feat(rocketmq): the instance resource supports 'tls_mode' field

docs/resources/dms_rocketmq_instance.md

@@ -78,7 +78,9 @@ The following arguments are supported:
   The description can contain a maximum of `1,024` characters.
 
 * `ssl_enable` - (Optional, Bool, ForceNew) Specifies whether the RocketMQ SASL_SSL is enabled. Defaults to **false**.
-  Changing this parameter will create a new resource.
+  Changing this parameter will create a new resource.  
+  If this parameter is set to **true**, `tls_mode` can be omitted or must be set to **SSL**.
+  If this parameter is set to **false**, `tls_mode` cannot be set to **SSL**.
 
 * `ipv6_enable` - (Optional, Bool, ForceNew) Specifies whether to support IPv6. Defaults to **false**.
   Changing this parameter will create a new resource.
@@ -113,6 +115,12 @@ The following arguments are supported:
 * `configs` - (Optional, List) Specifies the instance configs.
   The [configs](#dms_configs) structure is documented below.
 
+* `tls_mode` - (Optional, String) Specifies TLS mode of the instance.  
+  The valid values are as follows:
+  + **PLAINTEXT**
+  + **SSL**
+  + **PERMISSIVE**
+
 <a name="dms_configs"></a>
 The `configs` block supports:
 

huaweicloud/services/acceptance/rocketmq/resource_huaweicloud_dms_rocketmq_instance_test.go

@@ -71,6 +71,8 @@ func TestAccDmsRocketMQInstance_basic(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "storage_space", "500"),
 					resource.TestCheckResourceAttrPair(resourceName, "engine_version", "data.huaweicloud_dms_rocketmq_flavors.test", "versions.0"),
 					resource.TestCheckResourceAttrPair(resourceName, "flavor_id", "data.huaweicloud_dms_rocketmq_flavors.test", "flavors.0.id"),
+					resource.TestCheckResourceAttr(resourceName, "ssl_enable", "true"),
+					resource.TestCheckResourceAttr(resourceName, "tls_mode", "SSL"),
 				),
 			},
 			{
@@ -89,6 +91,8 @@ func TestAccDmsRocketMQInstance_basic(t *testing.T) {
 					resource.TestCheckResourceAttrPair(resourceName, "flavor_id", "data.huaweicloud_dms_rocketmq_flavors.test", "flavors.1.id"),
 					resource.TestCheckResourceAttr(resourceName, "configs.0.name", "fileReservedTime"),
 					resource.TestCheckResourceAttr(resourceName, "configs.0.value", "72"),
+					resource.TestCheckResourceAttr(resourceName, "ssl_enable", "false"),
+					resource.TestCheckResourceAttr(resourceName, "tls_mode", "PLAINTEXT"),
 				),
 			},
 			{
@@ -337,6 +341,7 @@ resource "huaweicloud_dms_rocketmq_instance" "test" {
   flavor_id         = local.flavor.id
   storage_space     = 500  
   broker_num        = 1
+  tls_mode          = "SSL"
 
   tags = {
     key1 = "value1"
@@ -381,6 +386,7 @@ resource "huaweicloud_dms_rocketmq_instance" "test" {
   flavor_id         = local.newFlavor.id
   storage_space     = 1200 
   broker_num        = 2
+  tls_mode          = "PLAINTEXT"
 
   configs {
     name  = "fileReservedTime"

huaweicloud/services/rocketmq/common.go

@@ -1,10 +1,15 @@
 package rocketmq
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
+	"strings"
+	"time"
 
 	"github.com/chnsz/golangsdk"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 
 	"github.com/huaweicloud/terraform-provider-huaweicloud/huaweicloud/utils"
 )
@@ -32,3 +37,57 @@ func handleMultiOperationsError(err error) (bool, error) {
 	}
 	return false, err
 }
+
+func getInstanceTaskById(client *golangsdk.ServiceClient, instanceId, taskId string) (interface{}, error) {
+	getTaskHttpUrl := "v2/{project_id}/instances/{instance_id}/tasks/{task_id}"
+	getTaskPath := client.Endpoint + getTaskHttpUrl
+	getTaskPath = strings.ReplaceAll(getTaskPath, "{project_id}", client.ProjectID)
+	getTaskPath = strings.ReplaceAll(getTaskPath, "{instance_id}", instanceId)
+	getTaskPath = strings.ReplaceAll(getTaskPath, "{task_id}", taskId)
+	opt := golangsdk.RequestOpts{
+		KeepResponseBody: true,
+	}
+	resp, err := client.Request("GET", getTaskPath, &opt)
+	if err != nil {
+		return nil, err
+	}
+
+	return utils.FlattenResponse(resp)
+}
+
+func refreshInstanceTaskStatus(client *golangsdk.ServiceClient, instanceID, taskId string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		respBody, err := getInstanceTaskById(client, instanceID, taskId)
+		if err != nil {
+			return respBody, "ERROR", err
+		}
+
+		// status options: DELETED, SUCCESS, EXECUTING, FAILED, CREATED
+		// CREATED means the task is in progress.
+		status := utils.PathSearch("tasks[0].status", respBody, "").(string)
+		if status == "FAILED" {
+			return respBody, "FAILED", fmt.Errorf("unexpect status (%s)", status)
+		}
+
+		if status == "SUCCESS" {
+			return respBody, "COMPLETED", nil
+		}
+
+		return "continue", "PENDING", nil
+	}
+}
+
+func waitForInstanceTaskStatusCompleted(ctx context.Context, client *golangsdk.ServiceClient, instanceId, taskId string,
+	timeout time.Duration) error {
+	stateConf := &resource.StateChangeConf{
+		Pending:      []string{"PENDING"},
+		Target:       []string{"COMPLETED"},
+		Refresh:      refreshInstanceTaskStatus(client, instanceId, taskId),
+		Timeout:      timeout,
+		Delay:        10 * time.Second,
+		PollInterval: 20 * time.Second,
+	}
+
+	_, err := stateConf.WaitForStateContext(ctx)
+	return err
+}

huaweicloud/services/rocketmq/resource_huaweicloud_dms_rocketmq_instance.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"net/http"
 	"strings"
 	"time"
 
@@ -38,6 +39,7 @@ type dmsError struct {
 // @API RocketMQ POST /v2/{engine}/{project_id}/instances/{instance_id}/extend
 // @API RocketMQ PUT /v2/{project_id}/rocketmq/instances/{instance_id}/configs
 // @API RocketMQ GET /v2/{project_id}/rocketmq/instances/{instance_id}/configs
+// @API RocketMQ POST /v2/{project_id}/{engine}/instances/{instance_id}/plain-ssl-switch
 // @API EIP GET /v1/{project_id}/publicips
 // @API BSS GET /v2/orders/customer-orders/details/{order_id}
 // @API BSS POST /v2/orders/subscriptions/resources/autorenew/{instance_id}
@@ -184,6 +186,14 @@ func ResourceDmsRocketMQInstance() *schema.Resource {
 				Computed:    true,
 				Description: `Specifies the instance configs.`,
 			},
+			// From the behavior of the Console page, this parameter is required.
+			// In order to ensure that the default value is returned in future, set Computed behavior.
+			"tls_mode": {
+				Type:        schema.TypeString,
+				Optional:    true,
+				Computed:    true,
+				Description: `The TLS mode of the instance.`,
+			},
 			"status": {
 				Type:        schema.TypeString,
 				Computed:    true,
@@ -506,6 +516,7 @@ func buildCreateRocketmqInstanceBodyParams(d *schema.ResourceData, cfg *config.C
 		"publicip_id":           utils.ValueIgnoreEmpty(d.Get("publicip_id")),
 		"broker_num":            utils.ValueIgnoreEmpty(d.Get("broker_num")),
 		"enterprise_project_id": utils.ValueIgnoreEmpty(cfg.GetEnterpriseProjectID(d)),
+		"tls_mode":              utils.ValueIgnoreEmpty(d.Get("tls_mode")),
 	}
 	if chargingMode, ok := d.GetOk("charging_mode"); ok && chargingMode == "prePaid" {
 		bodyParams["bss_param"] = buildCreateRocketmqInstanceBodyBssParams(d)
@@ -579,6 +590,46 @@ func buildRocketmqConfigsRequestBody(configs []interface{}) []map[string]string
 	return rst
 }
 
+func updateRocketmqInstanceTLSMode(ctx context.Context, client *golangsdk.ServiceClient, timeout time.Duration,
+	instanceId string, tlsMode string) error {
+	httpUrl := "v2/{project_id}/rocketmq/instances/{instance_id}/plain-ssl-switch"
+	updatePath := client.Endpoint + httpUrl
+	updatePath = strings.ReplaceAll(updatePath, "{project_id}", client.ProjectID)
+	updatePath = strings.ReplaceAll(updatePath, "{instance_id}", instanceId)
+
+	updateOpt := golangsdk.RequestOpts{
+		KeepResponseBody: true,
+		JSONBody: map[string]interface{}{
+			"tls_mode": tlsMode,
+		},
+	}
+
+	retryFunc := func() (interface{}, bool, error) {
+		resp, err := client.Request("POST", updatePath, &updateOpt)
+		retry, err := handleMultiOperationsError(err)
+		return resp, retry, err
+	}
+	resp, err := common.RetryContextWithWaitForState(&common.RetryContextWithWaitForStateParam{
+		Ctx:          ctx,
+		RetryFunc:    retryFunc,
+		WaitFunc:     rocketmqInstanceStateRefreshFunc(client, instanceId),
+		WaitTarget:   []string{"RUNNING"},
+		Timeout:      timeout,
+		DelayTimeout: 10 * time.Second,
+		PollInterval: 20 * time.Second,
+	})
+	if err != nil {
+		return err
+	}
+
+	respBody, err := utils.FlattenResponse(resp.(*http.Response))
+	if err != nil {
+		return err
+	}
+
+	return waitForInstanceTaskStatusCompleted(ctx, client, instanceId, utils.PathSearch("job_id", respBody, "").(string), timeout)
+}
+
 func resourceDmsRocketMQInstanceUpdate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	cfg := meta.(*config.Config)
 	region := cfg.GetRegion(d)
@@ -714,6 +765,14 @@ func resourceDmsRocketMQInstanceUpdate(ctx context.Context, d *schema.ResourceDa
 		}
 	}
 
+	if d.HasChange("tls_mode") {
+		err = updateRocketmqInstanceTLSMode(ctx, updateRocketmqInstanceClient, d.Timeout(schema.TimeoutUpdate),
+			instanceId, d.Get("tls_mode").(string))
+		if err != nil {
+			return diag.Errorf("error updating SSL mode of the RocketMQ instance (%s): %s", instanceId, err)
+		}
+	}
+
 	return resourceDmsRocketMQInstanceRead(ctx, d, meta)
 }
 
@@ -882,6 +941,7 @@ func resourceDmsRocketMQInstanceRead(_ context.Context, d *schema.ResourceData,
 		d.Set("resource_spec_code", utils.PathSearch("resource_spec_code", getRocketmqInstanceRespBody, nil)),
 		d.Set("cross_vpc_accesses", crossVpcAccess),
 		d.Set("charging_mode", chargingMode),
+		d.Set("tls_mode", utils.PathSearch("tls_mode", getRocketmqInstanceRespBody, nil)),
 	)
 
 	// get configs