astra_cdc: handle changing order of regions/tables (#461)

Change the CDC regions and tables from list to set to avoid issues with
order returned from rest API.
Change the data topic output to a map of maps for easier access to
specific topics.

Author: pgier

docs/resources/cdc_v3.md

@@ -39,7 +39,7 @@ resource "astra_streaming_tenant" "streaming_tenant2" {
 # Create a new database
 resource "astra_database" "db_database" {
   name                = "mydb"
-  keyspace            = "click_data"                # 48 characters max
+  keyspace            = "ks1"                       # 48 characters max
   cloud_provider      = "gcp"                       # this must match the cloud_provider of the streaming tenants
   regions             = ["us-central1", "us-east1"] # this must match the regions of the streaming tenants
   deletion_protection = false
@@ -50,7 +50,7 @@ resource "astra_table" "db_table1" {
   keyspace           = astra_database.db_database.keyspace
   database_id        = astra_database.db_database.id
   region             = astra_database.db_database.regions[0]
-  table              = "all_product_clicks"
+  table              = "table1"
   clustering_columns = "click_timestamp"
   partition_keys     = "visitor_id:click_url"
   column_definitions = [
@@ -84,19 +84,23 @@ resource "astra_cdc_v3" "db_cdc" {
   ]
   regions = [
     {
-      region            = "us-central-1"
+      region            = "us-central1"
       datacenter_id     = "${astra_database.db_database.id}-1"
       streaming_cluster = astra_streaming_tenant.streaming_tenant1.cluster_name
       streaming_tenant  = astra_streaming_tenant.streaming_tenant1.tenant_name
     },
     {
-      region            = "us-east-1"
+      region            = "us-east1"
       datacenter_id     = "${astra_database.db_database.id}-2"
       streaming_cluster = astra_streaming_tenant.streaming_tenant2.cluster_name
       streaming_tenant  = astra_streaming_tenant.streaming_tenant2.tenant_name
     },
   ]
 }
+
+output "streaming_data_topic1" {
+  value = astra_cdc_v3.example1.data_topics["us-east1"]["ks1.table1"]
+}
 ```
 
 <!-- schema generated by tfplugindocs -->
@@ -106,8 +110,12 @@ resource "astra_cdc_v3" "db_cdc" {
 
 - `database_id` (String) Astra database to create the keyspace.
 - `database_name` (String) Astra database name.
-- `regions` (Attributes List) Mapping between datacenter regions and streaming tenants. (see [below for nested schema](#nestedatt--regions))
-- `tables` (Attributes List) List of tables to enable CDC.  Must include at least 1. (see [below for nested schema](#nestedatt--tables))
+- `regions` (Attributes Set) Mapping between datacenter regions and streaming tenants. (see [below for nested schema](#nestedatt--regions))
+- `tables` (Attributes Set) List of tables to enable CDC.  Must include at least 1. (see [below for nested schema](#nestedatt--tables))
+
+### Read-Only
+
+- `data_topics` (Map of Map of String) Map of CDC data topics for each table in each region. Key is the region in the format `<region>`,
 
 <a id="nestedatt--regions"></a>
 ### Nested Schema for `regions`
@@ -128,10 +136,6 @@ Required:
 - `keyspace` (String)
 - `table` (String)
 
-Read-Only:
-
-- `data_topics` (List of String) List of Pulsar topics to which CDC data is published.  One data topic per region, in the same order of regions.
-
 ## Import
 
 Import is supported using the following syntax:

examples/resources/astra_cdc_v3/resource.tf

@@ -24,7 +24,7 @@ resource "astra_streaming_tenant" "streaming_tenant2" {
 # Create a new database
 resource "astra_database" "db_database" {
   name                = "mydb"
-  keyspace            = "click_data"                # 48 characters max
+  keyspace            = "ks1"                       # 48 characters max
   cloud_provider      = "gcp"                       # this must match the cloud_provider of the streaming tenants
   regions             = ["us-central1", "us-east1"] # this must match the regions of the streaming tenants
   deletion_protection = false
@@ -35,7 +35,7 @@ resource "astra_table" "db_table1" {
   keyspace           = astra_database.db_database.keyspace
   database_id        = astra_database.db_database.id
   region             = astra_database.db_database.regions[0]
-  table              = "all_product_clicks"
+  table              = "table1"
   clustering_columns = "click_timestamp"
   partition_keys     = "visitor_id:click_url"
   column_definitions = [
@@ -69,16 +69,20 @@ resource "astra_cdc_v3" "db_cdc" {
   ]
   regions = [
     {
-      region            = "us-central-1"
+      region            = "us-central1"
       datacenter_id     = "${astra_database.db_database.id}-1"
       streaming_cluster = astra_streaming_tenant.streaming_tenant1.cluster_name
       streaming_tenant  = astra_streaming_tenant.streaming_tenant1.tenant_name
     },
     {
-      region            = "us-east-1"
+      region            = "us-east1"
       datacenter_id     = "${astra_database.db_database.id}-2"
       streaming_cluster = astra_streaming_tenant.streaming_tenant2.cluster_name
       streaming_tenant  = astra_streaming_tenant.streaming_tenant2.tenant_name
     },
   ]
 }
+
+output "streaming_data_topic1" {
+  value = astra_cdc_v3.example1.data_topics["us-east1"]["ks1.table1"]
+}

internal/provider/resource_cdc_v3.go

@@ -39,18 +39,18 @@ func (r *CDCResource) Configure(_ context.Context, req resource.ConfigureRequest
 	r.clients = req.ProviderData.(*astraClients2)
 }
 
-// CDCResourceModel represents the resource data model
+// CDCResourceModel represents data used to configure CDC
 type CDCResourceModel struct {
 	DatabaseID   types.String               `tfsdk:"database_id"`
 	DatabaseName types.String               `tfsdk:"database_name"`
 	Tables       []KeyspaceTable            `tfsdk:"tables"`
 	Regions      []DatacenterToStreamingMap `tfsdk:"regions"`
+	DataTopics   types.Map                  `tfsdk:"data_topics"`
 }
 
 type KeyspaceTable struct {
-	Keyspace   types.String `tfsdk:"keyspace"`
-	Table      types.String `tfsdk:"table"`
-	DataTopics types.List   `tfsdk:"data_topics"`
+	Keyspace types.String `tfsdk:"keyspace"`
+	Table    types.String `tfsdk:"table"`
 }
 
 type DatacenterToStreamingMap struct {
@@ -60,20 +60,6 @@ type DatacenterToStreamingMap struct {
 	StreamingTenant  types.String `tfsdk:"streaming_tenant"`
 }
 
-// setDataTopics updates the data topics field for each table based on caculateCDCDataTopicName.
-func (m *CDCResourceModel) setDataTopics() {
-
-	for i := range m.Tables {
-		dataTopics := []attr.Value{}
-
-		for _, region := range m.Regions {
-			topicFQDN := calculateCDCDataTopicName(region.StreamingTenant.ValueString(), m.DatabaseID.ValueString(), m.Tables[i].Keyspace.ValueString(), m.Tables[i].Table.ValueString())
-			dataTopics = append(dataTopics, types.StringValue(topicFQDN))
-		}
-		m.Tables[i].DataTopics, _ = types.ListValue(types.StringType, dataTopics)
-	}
-}
-
 func (r *CDCResource) Metadata(_ context.Context, req resource.MetadataRequest, resp *resource.MetadataResponse) {
 	resp.TypeName = "astra_cdc_v3"
 }
@@ -93,7 +79,7 @@ func (r *CDCResource) Schema(_ context.Context, req resource.SchemaRequest, resp
 				Description: "Astra database name.",
 				Required:    true,
 			},
-			"tables": schema.ListNestedAttribute{
+			"tables": schema.SetNestedAttribute{
 				Description: "List of tables to enable CDC.  Must include at least 1.",
 				Required:    true,
 				NestedObject: schema.NestedAttributeObject{
@@ -104,17 +90,11 @@ func (r *CDCResource) Schema(_ context.Context, req resource.SchemaRequest, resp
 						"table": schema.StringAttribute{
 							Required: true,
 						},
-						"data_topics": schema.ListAttribute{
-							Description: "List of Pulsar topics to which CDC data is published.  " +
-								"One data topic per region, in the same order of regions.",
-							Computed:    true,
-							ElementType: types.StringType,
-						},
 					},
 				},
 			},
 
-			"regions": schema.ListNestedAttribute{
+			"regions": schema.SetNestedAttribute{
 				Description: "Mapping between datacenter regions and streaming tenants.",
 				Required:    true,
 				NestedObject: schema.NestedAttributeObject{
@@ -138,6 +118,14 @@ func (r *CDCResource) Schema(_ context.Context, req resource.SchemaRequest, resp
 					},
 				},
 			},
+			"data_topics": schema.MapAttribute{
+				Description: "Map of CDC data topics for each table in each region. " +
+					"Key is the region in the format `<region>`, ",
+				Computed: true,
+				ElementType: types.MapType{
+					ElemType: types.StringType,
+				},
+			},
 		},
 	}
 }
@@ -167,6 +155,7 @@ func (r *CDCResource) Create(ctx context.Context, req resource.CreateRequest, re
 	}
 
 	plan.setDataTopics()
+
 	diags = resp.State.Set(ctx, &plan)
 	resp.Diagnostics.Append(diags...)
 }
@@ -194,7 +183,9 @@ func (r *CDCResource) Read(ctx context.Context, req resource.ReadRequest, resp *
 		return
 	}
 
-	updateStateForCDCReadRequest(&state, cdcResponse.JSON200)
+	copyResponseDataToResourceState(&state, cdcResponse.JSON200)
+	state.setDataTopics()
+
 	diags = resp.State.Set(ctx, &state)
 	resp.Diagnostics.Append(diags...)
 }
@@ -287,7 +278,8 @@ func createEnableCDCRequestBody(tfData *CDCResourceModel) astra.EnableCDCJSONReq
 	return reqData
 }
 
-func updateStateForCDCReadRequest(tfData *CDCResourceModel, respData *astra.ListCDCResponse) {
+// copyResponseDataToResourceState copies the data from the REST endpoing response to the Terraform resource state model.
+func copyResponseDataToResourceState(tfData *CDCResourceModel, respData *astra.ListCDCResponse) {
 	tfData.DatabaseID = types.StringValue(respData.DatabaseID)
 	tfData.DatabaseName = types.StringValue(respData.DatabaseName)
 	var tables []KeyspaceTable
@@ -309,7 +301,6 @@ func updateStateForCDCReadRequest(tfData *CDCResourceModel, respData *astra.List
 		})
 	}
 	tfData.Regions = regions
-	tfData.setDataTopics()
 }
 
 func createDeleteCDCRequestBody(tfData *CDCResourceModel) astra.DeleteCDCJSONRequestBody {
@@ -333,3 +324,26 @@ const AstraCDCPulsarNamespace = "astracdc"
 func calculateCDCDataTopicName(streamingTenant, databaseID, keyspace, tableName string) string {
 	return fmt.Sprintf("persistent://%s/%s/data-%s-%s.%s", streamingTenant, AstraCDCPulsarNamespace, databaseID, keyspace, tableName)
 }
+
+// getDataTopicsList uses the region and table config to create the two dimensional (region and table) map of data topics.
+func (m *CDCResourceModel) setDataTopics() {
+
+	dataTopicsMap := map[string]attr.Value{}
+
+	for _, region := range m.Regions {
+		regionName := region.Region.ValueString()
+		regionDataTopics := make(map[string]attr.Value)
+
+		for _, table := range m.Tables {
+			keyspaceTable := fmt.Sprintf("%s.%s", table.Keyspace.ValueString(), table.Table.ValueString())
+			topicFQDN := calculateCDCDataTopicName(region.StreamingTenant.ValueString(), m.DatabaseID.ValueString(), table.Keyspace.ValueString(), table.Table.ValueString())
+			regionDataTopics[keyspaceTable] = types.StringValue(topicFQDN)
+		}
+		dataTopicsMap[regionName] = types.MapValueMust(
+			types.StringType,
+			regionDataTopics,
+		)
+	}
+
+	m.DataTopics = types.MapValueMust(types.MapType{ElemType: types.StringType}, dataTopicsMap)
+}