Skip to content

feat/tdmq #2451

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Jan 5, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .changelog/2451.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
```release-note:new-resource
tencentcloud_tdmq_subscription
```
2 changes: 1 addition & 1 deletion tencentcloud/provider.go
Original file line number Diff line number Diff line change
Expand Up @@ -1463,12 +1463,12 @@ func Provider() *schema.Provider {
"tencentcloud_tdmq_topic": tpulsar.ResourceTencentCloudTdmqTopic(),
"tencentcloud_tdmq_role": tpulsar.ResourceTencentCloudTdmqRole(),
"tencentcloud_tdmq_namespace_role_attachment": tpulsar.ResourceTencentCloudTdmqNamespaceRoleAttachment(),
"tencentcloud_tdmq_subscription_attachment": tpulsar.ResourceTencentCloudTdmqSubscriptionAttachment(),
"tencentcloud_tdmq_rabbitmq_user": trabbit.ResourceTencentCloudTdmqRabbitmqUser(),
"tencentcloud_tdmq_rabbitmq_virtual_host": trabbit.ResourceTencentCloudTdmqRabbitmqVirtualHost(),
"tencentcloud_tdmq_rabbitmq_vip_instance": trabbit.ResourceTencentCloudTdmqRabbitmqVipInstance(),
"tencentcloud_tdmq_send_rocketmq_message": trocket.ResourceTencentCloudTdmqSendRocketmqMessage(),
"tencentcloud_tdmq_professional_cluster": tpulsar.ResourceTencentCloudTdmqProfessionalCluster(),
"tencentcloud_tdmq_subscription": tpulsar.ResourceTencentCloudTdmqSubscription(),
"tencentcloud_cos_bucket_policy": cos.ResourceTencentCloudCosBucketPolicy(),
"tencentcloud_cos_bucket_domain_certificate_attachment": cos.ResourceTencentCloudCosBucketDomainCertificateAttachment(),
"tencentcloud_cos_bucket_inventory": cos.ResourceTencentCloudCosBucketInventory(),
Expand Down
1 change: 1 addition & 0 deletions tencentcloud/provider.md
Original file line number Diff line number Diff line change
Expand Up @@ -683,6 +683,7 @@ TDMQ for Pulsar(tpulsar)
tencentcloud_tdmq_topic
tencentcloud_tdmq_role
tencentcloud_tdmq_namespace_role_attachment
tencentcloud_tdmq_subscription

TencentDB for MongoDB(mongodb)
Data Source
Expand Down
66 changes: 66 additions & 0 deletions tencentcloud/services/tdmq/service_tencentcloud_tdmq.go
Original file line number Diff line number Diff line change
Expand Up @@ -920,6 +920,39 @@ func (me *TdmqService) DescribeTdmqSubscriptionAttachmentById(ctx context.Contex
return
}

func (me *TdmqService) DescribeTdmqSubscriptionById(ctx context.Context, clusterId, environmentId, topicName, subscriptionName string) (subscriptionAttachment *tdmq.Subscription, errRet error) {
logId := tccommon.GetLogId(ctx)

request := tdmq.NewDescribeSubscriptionsRequest()
request.ClusterId = &clusterId
request.EnvironmentId = &environmentId
request.TopicName = &topicName
request.SubscriptionName = &subscriptionName

defer func() {
if errRet != nil {
log.Printf("[CRITAL]%s api[%s] fail, request body [%s], reason[%s]\n", logId, request.GetAction(), request.ToJsonString(), errRet.Error())
}
}()

ratelimit.Check(request.GetAction())

response, err := me.client.UseTdmqClient().DescribeSubscriptions(request)
if err != nil {
errRet = err
return
}

log.Printf("[DEBUG]%s api[%s] success, request body [%s], response body [%s]\n", logId, request.GetAction(), request.ToJsonString(), response.ToJsonString())

if *response.Response.TotalCount == 0 {
return
}

subscriptionAttachment = response.Response.SubscriptionSets[0]
return
}

func (me *TdmqService) GetTdmqTopicsAttachmentById(ctx context.Context, environmentId, Topic, subscriptionName, clusterId string) (has bool, errRet error) {
logId := tccommon.GetLogId(ctx)

Expand Down Expand Up @@ -1032,6 +1065,39 @@ func (me *TdmqService) DeleteTdmqSubscriptionAttachmentById(ctx context.Context,
return
}

func (me *TdmqService) DeleteTdmqSubscriptionById(ctx context.Context, clusterId, environmentId, topicName, subscriptionName string) (errRet error) {
logId := tccommon.GetLogId(ctx)

request := tdmq.NewDeleteSubscriptionsRequest()
request.SubscriptionTopicSets = []*tdmq.SubscriptionTopic{
{
EnvironmentId: &environmentId,
TopicName: &topicName,
SubscriptionName: &subscriptionName,
},
}
request.ClusterId = &clusterId
request.EnvironmentId = &environmentId

defer func() {
if errRet != nil {
log.Printf("[CRITAL]%s api[%s] fail, request body [%s], reason[%s]\n", logId, request.GetAction(), request.ToJsonString(), errRet.Error())
}
}()

ratelimit.Check(request.GetAction())

response, err := me.client.UseTdmqClient().DeleteSubscriptions(request)
if err != nil {
errRet = err
return
}

log.Printf("[DEBUG]%s api[%s] success, request body [%s], response body [%s]\n", logId, request.GetAction(), request.ToJsonString(), response.ToJsonString())

return
}

func (me *TdmqService) DescribeTdmqDeadLetterSourceQueueByFilter(ctx context.Context, param map[string]interface{}) (deadLetterSourceQueue []*tdmq.CmqDeadLetterSource, errRet error) {
var (
logId = tccommon.GetLogId(ctx)
Expand Down
249 changes: 249 additions & 0 deletions tencentcloud/services/tpulsar/resource_tc_tdmq_subscription.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,249 @@
package tpulsar

import (
"context"
"errors"
"fmt"
"log"
"strings"

svctdmq "github.com/tencentcloudstack/terraform-provider-tencentcloud/tencentcloud/services/tdmq"

"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
tdmq "github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/tdmq/v20200217"

tccommon "github.com/tencentcloudstack/terraform-provider-tencentcloud/tencentcloud/common"
"github.com/tencentcloudstack/terraform-provider-tencentcloud/tencentcloud/internal/helper"
)

func ResourceTencentCloudTdmqSubscription() *schema.Resource {
return &schema.Resource{
Create: resourceTencentCloudTdmqSubscriptionCreate,
Read: resourceTencentCloudTdmqSubscriptionRead,
Delete: resourceTencentCloudTdmqSubscriptionDelete,
Importer: &schema.ResourceImporter{
State: schema.ImportStatePassthrough,
},
Schema: map[string]*schema.Schema{
"cluster_id": {
Required: true,
ForceNew: true,
Type: schema.TypeString,
Description: "Pulsar cluster ID.",
},
"environment_id": {
Required: true,
ForceNew: true,
Type: schema.TypeString,
Description: "Environment (namespace) name.",
},
"topic_name": {
Required: true,
ForceNew: true,
Type: schema.TypeString,
Description: "Topic name.",
},
"subscription_name": {
Required: true,
ForceNew: true,
Type: schema.TypeString,
Description: "Subscriber name, which can contain up to 128 characters.",
},
"remark": {
Optional: true,
ForceNew: true,
Type: schema.TypeString,
Description: "Remarks (up to 128 characters).",
},
"auto_create_policy_topic": {
Optional: true,
ForceNew: true,
Type: schema.TypeBool,
Default: false,
Description: "Whether to automatically create a dead letter topic and a retry letter topic. true: yes; false: no(default value).",
},
"auto_delete_policy_topic": {
Optional: true,
ForceNew: true,
Type: schema.TypeBool,
Default: false,
Description: "Whether to automatically delete a dead letter topic and a retry letter topic. Setting is only allowed when `auto_create_policy_topic` is true. Default is false.",
},
},
}
}

func resourceTencentCloudTdmqSubscriptionCreate(d *schema.ResourceData, meta interface{}) error {
defer tccommon.LogElapsed("resource.tencentcloud_tdmq_subscription.create")()
defer tccommon.InconsistentCheck(d, meta)()

var (
logId = tccommon.GetLogId(tccommon.ContextNil)
request = tdmq.NewCreateSubscriptionRequest()
clusterId string
environmentId string
topicName string
subscriptionName string
autoCreatePolicyTopic bool
autoDeletePolicyTopic string
)

if v, ok := d.GetOk("cluster_id"); ok {
request.ClusterId = helper.String(v.(string))
clusterId = v.(string)
}

if v, ok := d.GetOk("environment_id"); ok {
request.EnvironmentId = helper.String(v.(string))
environmentId = v.(string)
}

if v, ok := d.GetOk("topic_name"); ok {
request.TopicName = helper.String(v.(string))
topicName = v.(string)
}

if v, ok := d.GetOk("subscription_name"); ok {
request.SubscriptionName = helper.String(v.(string))
subscriptionName = v.(string)
}

if v, ok := d.GetOk("remark"); ok {
request.Remark = helper.String(v.(string))
}

if v, ok := d.GetOkExists("auto_create_policy_topic"); ok {
request.AutoCreatePolicyTopic = helper.Bool(v.(bool))
autoCreatePolicyTopic = v.(bool)

if v, ok = d.GetOkExists("auto_delete_policy_topic"); ok {
if !autoCreatePolicyTopic && v.(bool) {
return errors.New("If `auto_create_policy_topic` is false, Can't set `auto_delete_policy_topic` param.")
} else {
if v.(bool) {
autoDeletePolicyTopic = "true"
} else {
autoDeletePolicyTopic = "false"
}
}
}
}

request.IsIdempotent = helper.Bool(false)

err := resource.Retry(tccommon.WriteRetryTimeout, func() *resource.RetryError {
result, e := meta.(tccommon.ProviderMeta).GetAPIV3Conn().UseTdmqClient().CreateSubscription(request)
if e != nil {
return tccommon.RetryError(e)
} else {
log.Printf("[DEBUG]%s api[%s] success, request body [%s], response body [%s]\n", logId, request.GetAction(), request.ToJsonString(), result.ToJsonString())
}

if result == nil || !*result.Response.Result {
e = fmt.Errorf("create tdmq subscription failed")
return resource.NonRetryableError(e)
}

return nil
})

if err != nil {
log.Printf("[CRITAL]%s create tdmq subscription failed, reason:%+v", logId, err)
return err
}

d.SetId(strings.Join([]string{clusterId, environmentId, topicName, subscriptionName, autoDeletePolicyTopic}, tccommon.FILED_SP))

return resourceTencentCloudTdmqSubscriptionRead(d, meta)
}

func resourceTencentCloudTdmqSubscriptionRead(d *schema.ResourceData, meta interface{}) error {
defer tccommon.LogElapsed("resource.tencentcloud_tdmq_subscription.read")()
defer tccommon.InconsistentCheck(d, meta)()

var (
logId = tccommon.GetLogId(tccommon.ContextNil)
ctx = context.WithValue(context.TODO(), tccommon.LogIdKey, logId)
service = svctdmq.NewTdmqService(meta.(tccommon.ProviderMeta).GetAPIV3Conn())
)

idSplit := strings.Split(d.Id(), tccommon.FILED_SP)

if len(idSplit) != 5 {
return fmt.Errorf("id is broken,%s", d.Id())
}

clusterId := idSplit[0]
environmentId := idSplit[1]
topicName := idSplit[2]
subscriptionName := idSplit[3]
autoDeletePolicyTopic := idSplit[4]

subscription, err := service.DescribeTdmqSubscriptionById(ctx, clusterId, environmentId, topicName, subscriptionName)
if err != nil {
return err
}

if subscription == nil {
d.SetId("")
log.Printf("[WARN]%s resource `TdmqSubscription` [%s] not found, please check if it has been deleted.\n", logId, d.Id())
return nil
}

_ = d.Set("environment_id", subscription.EnvironmentId)
_ = d.Set("topic_name", subscription.TopicName)
_ = d.Set("subscription_name", subscriptionName)
_ = d.Set("cluster_id", clusterId)
_ = d.Set("remark", subscription.Remark)

if autoDeletePolicyTopic == "true" {
_ = d.Set("auto_delete_policy_topic", true)
} else {
_ = d.Set("auto_delete_policy_topic", false)
}

// Get Topics Status For auto_create_policy_topic
has, err := service.GetTdmqTopicsAttachmentById(ctx, environmentId, topicName, subscriptionName, clusterId)
if err != nil {
return err
}

_ = d.Set("auto_create_policy_topic", has)
return nil
}

func resourceTencentCloudTdmqSubscriptionDelete(d *schema.ResourceData, meta interface{}) error {
defer tccommon.LogElapsed("resource.tencentcloud_tdmq_subscription.delete")()
defer tccommon.InconsistentCheck(d, meta)()

var (
logId = tccommon.GetLogId(tccommon.ContextNil)
ctx = context.WithValue(context.TODO(), tccommon.LogIdKey, logId)
service = svctdmq.NewTdmqService(meta.(tccommon.ProviderMeta).GetAPIV3Conn())
)

idSplit := strings.Split(d.Id(), tccommon.FILED_SP)
if len(idSplit) != 5 {
return fmt.Errorf("id is broken,%s", d.Id())
}

clusterId := idSplit[0]
environmentId := idSplit[1]
topicName := idSplit[2]
subscriptionName := idSplit[3]
autoDeletePolicyTopic := idSplit[4]

if err := service.DeleteTdmqSubscriptionById(ctx, clusterId, environmentId, topicName, subscriptionName); err != nil {
return err
}

if autoDeletePolicyTopic == "true" {
// Delete Topics
if err := service.DeleteTdmqTopicsAttachmentById(ctx, environmentId, topicName, subscriptionName, clusterId); err != nil {
return err
}
}

return nil
}
Loading