Pub/Sub Geo Restriction (#4131)

Signed-off-by: Modular Magician <[email protected]>

Author: modular-magician

google/resource_pubsub_topic.go

@@ -58,6 +58,22 @@ func resourcePubsubTopic() *schema.Resource {
 				Optional: true,
 				Elem:     &schema.Schema{Type: schema.TypeString},
 			},
+			"message_storage_policy": {
+				Type:     schema.TypeList,
+				Optional: true,
+				MaxItems: 1,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"allowed_persistence_regions": {
+							Type:     schema.TypeList,
+							Required: true,
+							Elem: &schema.Schema{
+								Type: schema.TypeString,
+							},
+						},
+					},
+				},
+			},
 			"project": {
 				Type:     schema.TypeString,
 				Optional: true,
@@ -90,6 +106,12 @@ func resourcePubsubTopicCreate(d *schema.ResourceData, meta interface{}) error {
 	} else if v, ok := d.GetOkExists("labels"); !isEmptyValue(reflect.ValueOf(labelsProp)) && (ok || !reflect.DeepEqual(v, labelsProp)) {
 		obj["labels"] = labelsProp
 	}
+	messageStoragePolicyProp, err := expandPubsubTopicMessageStoragePolicy(d.Get("message_storage_policy"), d, config)
+	if err != nil {
+		return err
+	} else if v, ok := d.GetOkExists("message_storage_policy"); !isEmptyValue(reflect.ValueOf(messageStoragePolicyProp)) && (ok || !reflect.DeepEqual(v, messageStoragePolicyProp)) {
+		obj["messageStoragePolicy"] = messageStoragePolicyProp
+	}
 
 	obj, err = resourcePubsubTopicEncoder(d, meta, obj)
 	if err != nil {
@@ -149,6 +171,9 @@ func resourcePubsubTopicRead(d *schema.ResourceData, meta interface{}) error {
 	if err := d.Set("labels", flattenPubsubTopicLabels(res["labels"], d)); err != nil {
 		return fmt.Errorf("Error reading Topic: %s", err)
 	}
+	if err := d.Set("message_storage_policy", flattenPubsubTopicMessageStoragePolicy(res["messageStoragePolicy"], d)); err != nil {
+		return fmt.Errorf("Error reading Topic: %s", err)
+	}
 
 	return nil
 }
@@ -163,6 +188,12 @@ func resourcePubsubTopicUpdate(d *schema.ResourceData, meta interface{}) error {
 	} else if v, ok := d.GetOkExists("labels"); !isEmptyValue(reflect.ValueOf(v)) && (ok || !reflect.DeepEqual(v, labelsProp)) {
 		obj["labels"] = labelsProp
 	}
+	messageStoragePolicyProp, err := expandPubsubTopicMessageStoragePolicy(d.Get("message_storage_policy"), d, config)
+	if err != nil {
+		return err
+	} else if v, ok := d.GetOkExists("message_storage_policy"); !isEmptyValue(reflect.ValueOf(v)) && (ok || !reflect.DeepEqual(v, messageStoragePolicyProp)) {
+		obj["messageStoragePolicy"] = messageStoragePolicyProp
+	}
 
 	obj, err = resourcePubsubTopicUpdateEncoder(d, meta, obj)
 	if err != nil {
@@ -180,6 +211,10 @@ func resourcePubsubTopicUpdate(d *schema.ResourceData, meta interface{}) error {
 	if d.HasChange("labels") {
 		updateMask = append(updateMask, "labels")
 	}
+
+	if d.HasChange("message_storage_policy") {
+		updateMask = append(updateMask, "messageStoragePolicy")
+	}
 	// updateMask is a URL parameter but not present in the schema, so replaceVars
 	// won't set it
 	url, err = addQueryParams(url, map[string]string{"updateMask": strings.Join(updateMask, ",")})
@@ -249,6 +284,23 @@ func flattenPubsubTopicLabels(v interface{}, d *schema.ResourceData) interface{}
 	return v
 }
 
+func flattenPubsubTopicMessageStoragePolicy(v interface{}, d *schema.ResourceData) interface{} {
+	if v == nil {
+		return nil
+	}
+	original := v.(map[string]interface{})
+	if len(original) == 0 {
+		return nil
+	}
+	transformed := make(map[string]interface{})
+	transformed["allowed_persistence_regions"] =
+		flattenPubsubTopicMessageStoragePolicyAllowedPersistenceRegions(original["allowedPersistenceRegions"], d)
+	return []interface{}{transformed}
+}
+func flattenPubsubTopicMessageStoragePolicyAllowedPersistenceRegions(v interface{}, d *schema.ResourceData) interface{} {
+	return v
+}
+
 func expandPubsubTopicName(v interface{}, d TerraformResourceData, config *Config) (interface{}, error) {
 	return GetResourceNameFromSelfLink(v.(string)), nil
 }
@@ -268,6 +320,29 @@ func expandPubsubTopicLabels(v interface{}, d TerraformResourceData, config *Con
 	return m, nil
 }
 
+func expandPubsubTopicMessageStoragePolicy(v interface{}, d TerraformResourceData, config *Config) (interface{}, error) {
+	l := v.([]interface{})
+	if len(l) == 0 || l[0] == nil {
+		return nil, nil
+	}
+	raw := l[0]
+	original := raw.(map[string]interface{})
+	transformed := make(map[string]interface{})
+
+	transformedAllowedPersistenceRegions, err := expandPubsubTopicMessageStoragePolicyAllowedPersistenceRegions(original["allowed_persistence_regions"], d, config)
+	if err != nil {
+		return nil, err
+	} else if val := reflect.ValueOf(transformedAllowedPersistenceRegions); val.IsValid() && !isEmptyValue(val) {
+		transformed["allowedPersistenceRegions"] = transformedAllowedPersistenceRegions
+	}
+
+	return transformed, nil
+}
+
+func expandPubsubTopicMessageStoragePolicyAllowedPersistenceRegions(v interface{}, d TerraformResourceData, config *Config) (interface{}, error) {
+	return v, nil
+}
+
 func resourcePubsubTopicEncoder(d *schema.ResourceData, meta interface{}, obj map[string]interface{}) (map[string]interface{}, error) {
 	delete(obj, "name")
 	return obj, nil

google/resource_pubsub_topic_generated_test.go

@@ -60,6 +60,45 @@ resource "google_pubsub_topic" "example" {
 `, context)
 }
 
+func TestAccPubsubTopic_pubsubTopicGeoRestrictedExample(t *testing.T) {
+	t.Parallel()
+
+	context := map[string]interface{}{
+		"random_suffix": acctest.RandString(10),
+	}
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckPubsubTopicDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccPubsubTopic_pubsubTopicGeoRestrictedExample(context),
+			},
+			{
+				ResourceName:      "google_pubsub_topic.example",
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func testAccPubsubTopic_pubsubTopicGeoRestrictedExample(context map[string]interface{}) string {
+	return Nprintf(`
+resource "google_pubsub_topic" "example" {
+  name = "example-topic%{random_suffix}"
+
+  message_storage_policy {
+    allowed_persistence_regions = [
+      "europe-west3",
+    ]
+  }
+
+}
+`, context)
+}
+
 func testAccCheckPubsubTopicDestroy(s *terraform.State) error {
 	for name, rs := range s.RootModule().Resources {
 		if rs.Type != "google_pubsub_topic" {

google/resource_pubsub_topic_test.go

@@ -28,7 +28,7 @@ func TestAccPubsubTopic_update(t *testing.T) {
 				ImportStateVerify: true,
 			},
 			{
-				Config: testAccPubsubTopic_update(topic, "wibble", "wobble"),
+				Config: testAccPubsubTopic_updateWithRegion(topic, "wibble", "wobble", "us-central1"),
 			},
 			{
 				ResourceName:      "google_pubsub_topic.foo",
@@ -74,6 +74,23 @@ resource "google_pubsub_topic" "foo" {
 `, topic, key, value)
 }
 
+func testAccPubsubTopic_updateWithRegion(topic, key, value, region string) string {
+	return fmt.Sprintf(`
+resource "google_pubsub_topic" "foo" {
+	name = "%s"
+	labels = {
+		%s = "%s"
+	}
+
+	message_storage_policy {
+		allowed_persistence_regions = [
+		  "%s",
+		]
+	}
+}
+`, topic, key, value, region)
+}
+
 func testAccPubsubTopic_cmek(pid, topicName, kmsKey string) string {
 	return fmt.Sprintf(`
 data "google_project" "project" {

website/docs/r/pubsub_topic.html.markdown

@@ -66,6 +66,26 @@ resource "google_kms_key_ring" "key_ring" {
   location = "global"
 }
 ```
+<div class = "oics-button" style="float: right; margin: 0 0 -15px">
+  <a href="https://console.cloud.google.com/cloudshell/open?cloudshell_git_repo=https%3A%2F%2Fgithub.jpy.wang%2Fterraform-google-modules%2Fdocs-examples.git&cloudshell_working_dir=pubsub_topic_geo_restricted&cloudshell_image=gcr.io%2Fgraphite-cloud-shell-images%2Fterraform%3Alatest&open_in_editor=main.tf&cloudshell_print=.%2Fmotd&cloudshell_tutorial=.%2Ftutorial.md" target="_blank">
+    <img alt="Open in Cloud Shell" src="//gstatic.com/cloudssh/images/open-btn.svg" style="max-height: 44px; margin: 32px auto; max-width: 100%;">
+  </a>
+</div>
+## Example Usage - Pubsub Topic Geo Restricted
+
+
+```hcl
+resource "google_pubsub_topic" "example" {
+  name = "example-topic"
+
+  message_storage_policy {
+    allowed_persistence_regions = [
+      "europe-west3",
+    ]
+  }
+
+}
+```
 
 ## Argument Reference
 
@@ -92,10 +112,27 @@ The following arguments are supported:
   (Optional)
   A set of key/value label pairs to assign to this Topic.
 
+* `message_storage_policy` -
+  (Optional)
+  Policy constraining the set of Google Cloud Platform regions where
+  messages published to the topic may be stored. If not present, then no
+  constraints are in effect.  Structure is documented below.
+
 * `project` - (Optional) The ID of the project in which the resource belongs.
     If it is not provided, the provider project is used.
 
 
+The `message_storage_policy` block supports:
+
+* `allowed_persistence_regions` -
+  (Required)
+  A list of IDs of GCP regions where messages that are published to
+  the topic may be persisted in storage. Messages published by
+  publishers running in non-allowed GCP regions (or running outside
+  of GCP altogether) will be routed for storage in one of the
+  allowed regions. An empty list means that no regions are allowed,
+  and is not a valid configuration.
+
 
 ## Timeouts