全部產品
Search

搜尋本產品

    Resource Management：使用OOS將RocketMQ執行個體標籤繼承到Topic和Group

    文件中心

    Resource Management：使用OOS將RocketMQ執行個體標籤繼承到Topic和Group

    更新時間：Dec 04, 2025

    通過建立和執行系統營運管理（OOS）自訂模板，您可以將RocketMQ執行個體的標籤繼承到執行個體中的Topic和Group上。

    使用限制

    預設情況下，某地區的OOS負責管理本地區的資源。例如，華東1（杭州）的OOS預設管理華東1（杭州）的ECS執行個體。但是，作為例外，使用者可以在模板ExecuteAPI裡指定RegionId的值，來調用其他地區的OpenAPI（不建議這樣做）。因此，此處OOS的地區需要與RocketMQ執行個體地區保持一致。關於OOS的使用限制，請參見使用限制。

    操作步驟

    1. 登入OOS控制台。

    2. 建立子模板Ons_topicExtendInstanceTags。

      模板樣本，請參見子模板：根據執行個體ID及其他條件更新Topic的標籤。建立模板的具體操作，請參見建立模板。

    3. 建立子模板Ons_groupExtendInstanceTags。

      模板樣本，請參見子模板：根據執行個體ID及輸入條件更新Group的標籤。建立模板的具體操作，請參見建立模板。

    4. 建立父模板RocketMQInstanceTagInherit，父模板中會嵌套以上兩個子模板。

    5. 對父模板RocketMQInstanceTagInherit建立執行。

      您需要配置以下資訊：

      • 選擇需要繼承標籤的地區。

      • 選擇需要繼承的標籤。

      • 選擇需要繼承標籤的資源類型。

      • 選擇相同標籤鍵不同標籤值時是否需要覆蓋。

      建立執行的具體操作，請參見執行概述。

    6. 登入訊息佇列RocketMQ版控制台，驗證RocketMQ執行個體中的Topic和Group是否繼承了執行個體的對應標籤。

    子模板：根據執行個體ID及其他條件更新Topic的標籤

    模板名稱：Ons_topicExtendInstanceTags

    模板內容：

    FormatVersion: OOS-2019-06-01
Description: 標籤繼承
Parameters:
  regionId:
    Type: String
    Description:
      en: The id of region
      zh-cn: 地區ID
    Label:
      en: Region
      zh-cn: 地區
    AssociationProperty: RegionId
    Default: '{{ ACS::RegionId }}'
  instanceId:
    Type: String
    Description:
      en: The id of OnsInstance
      zh-cn: RocketMQ執行個體ID
    Label:
      en: InstanceId
      zh-cn: RocketMQ執行個體ID
  tagKeys:
    Type: List
    Description: 標籤Key的集合
  isUpdate:
    Type: Boolean
    Description: 是否覆蓋
    Default: false
RamRole: ''
Tasks:
  - Name: getOnsInstanceAndTags
    Action: ACS::ExecuteAPI
    Description: ListTagResources
    Properties:
      Service: ONS
      API: ListTagResources
      Parameters:
        RegionId: '{{ regionId }}'
        ResourceId:
          - '{{instanceId}}'
        ResourceType: instance
    Outputs:
      onsTags:
        Type: Json
        ValueSelector: '.TagResources | map(select( .TagKey | test("^(?!acs).*"))) | map(select(.TagKey as $tagKey | {{tagKeys}} | index($tagKey) >=0)) '
  - Name: getTopicAndTags
    Action: ACS::ExecuteAPI
    Description: OnsTopicList
    Properties:
      Service: Ons
      API: OnsTopicList
      Parameters:
        RegionId: '{{ regionId }}'
        InstanceId: '{{instanceId}}'
    Outputs:
      updateTopicAndTags:
        Type: List
        ValueSelector: .Data.PublishInfoDo[] | {"Topic":.Topic, "Tags":.Tags}
  - Name: tagResources
    Action: ACS::ExecuteAPI
    Description: TagTopics
    Properties:
      Service: Ons
      API: TagResources
      Parameters:
        RegionId: '{{ regionId }}'
        ResourceId:
          Fn::Jq:
            - All
            - .Topic
            - '{{ACS::TaskLoopItem}}'
        Tag:
          Fn::If:
            - '{{isUpdate}}'
            - Fn::Jq:
                - All
                - '.[] |{"Key":.TagKey, "Value": .TagValue} '
                - '{{getOnsInstanceAndTags.onsTags}}'
            - Fn::Jq:
                - All
                - '.Tags.Tag as $topicTags | {{getOnsInstanceAndTags.onsTags}} | map(select(.TagKey as $tagKey | [$topicTags[].Key] | index($tagKey) <0)) |.[] |{"Key":.TagKey, "Value": .TagValue}'
                - '{{ACS::TaskLoopItem}}'
        ResourceType: topic
        InstanceId: '{{instanceId}}'
    Outputs:
      reqResult:
        Type: Json
        ValueSelector: .RequestId
    Loop:
      RateControl:
        Mode: Concurrency
        MaxErrors: 100
        Concurrency: 1
      Items: '{{getTopicAndTags.updateTopicAndTags}}'
      Outputs:
        tagResult:
          AggregateType: Fn::ListJoin
          AggregateField: reqResult

    子模板：根據執行個體ID及輸入條件更新Group的標籤

    模板名稱：Ons_groupExtendInstanceTags

    模板內容：

    FormatVersion: OOS-2019-06-01
Description: 標籤繼承
Parameters:
  regionId:
    Type: String
    Description:
      en: The id of region
      zh-cn: 地區ID
    Label:
      en: Region
      zh-cn: 地區
    AssociationProperty: RegionId
    Default: '{{ ACS::RegionId }}'
  instanceId:
    Type: String
    Description:
      en: The id of OnsInstance
      zh-cn: 執行個體ID
    Label:
      en: InstanceId
      zh-cn: 執行個體ID
  tagKeys:
    Type: List
    Description: 標籤Key的集合
  isUpdate:
    Type: Boolean
    Description: 是否覆蓋
    Default: false
RamRole: ''
Tasks:
  - Name: getOnsInstanceAndTags
    Action: ACS::ExecuteAPI
    Description: ListTagResources
    Properties:
      Service: ONS
      API: ListTagResources
      Parameters:
        RegionId: '{{ regionId }}'
        ResourceId:
          - '{{instanceId}}'
        ResourceType: instance
    Outputs:
      onsTags:
        Type: Json
        ValueSelector: '.TagResources | map(select( .TagKey | test("^(?!acs).*"))) | map(select(.TagKey as $tagKey | {{tagKeys}} | index($tagKey) >=0)) '
  - Name: getGroupAndTags
    Action: ACS::ExecuteAPI
    Description: OnsGroupList
    Properties:
      Service: Ons
      API: OnsGroupList
      Parameters:
        RegionId: '{{ regionId }}'
        InstanceId: '{{instanceId}}'
    Outputs:
      updateGroupAndTags:
        Type: List
        ValueSelector: .Data.SubscribeInfoDo[] | {"Group":.GroupId, "Tags":.Tags}
  - Name: tagResources
    Action: ACS::ExecuteAPI
    Description: TagGroups
    Properties:
      Service: Ons
      API: TagResources
      Parameters:
        RegionId: '{{ regionId }}'
        ResourceId:
          Fn::Jq:
            - All
            - .Group
            - '{{ACS::TaskLoopItem}}'
        Tag:
          Fn::If:
            - '{{isUpdate}}'
            - Fn::Jq:
                - All
                - '.[] |{"Key":.TagKey, "Value": .TagValue} '
                - '{{getOnsInstanceAndTags.onsTags}}'
            - Fn::Jq:
                - All
                - '.Tags.Tag as $groupTags | {{getOnsInstanceAndTags.onsTags}} | map(select(.TagKey as $tagKey | [$groupTags[].Key] | index($tagKey) <0)) |.[] |{"Key":.TagKey, "Value": .TagValue}'
                - '{{ACS::TaskLoopItem}}'
        ResourceType: group
        InstanceId: '{{instanceId}}'
    Outputs:
      reqResult:
        Type: Json
        ValueSelector: .RequestId
    Loop:
      RateControl:
        Mode: Concurrency
        MaxErrors: 100
        Concurrency: 1
      Items: '{{getGroupAndTags.updateGroupAndTags}}'
      Outputs:
        tagResult:
          AggregateType: Fn::ListJoin
          AggregateField: reqResult

    父模板：按照輸入條件更新某個地區下所有RocketMQ執行個體中的Topic和Group的標籤

    模板名稱：RocketMQInstanceTagInherit

    模板內容：

    FormatVersion: OOS-2019-06-01
Description: 標籤繼承
Parameters:
  regionIds:
    Type: List
    Description:
      en: The id of region
      zh-cn: 地區ID
    Label:
      en: Region
      zh-cn: 地區
    AssociationProperty: RegionId
  tags:
    Type: List
    AssociationProperty: Tags
    AssociationPropertyMetadata:
      ResourceType: ALIYUN::MQ::INSTANCE
    Description: 訊息佇列RocketMQ執行個體上的標籤，選擇需要繼承到Topic或Group上的標籤鍵及標籤值。
  resourceTypes:
    Type: List
    Description: 所選擇繼承的資源類型
    AllowedValues:
      - topic
      - group
  isUpdate:
    Type: Boolean
    Description: 如果被繼承的資源存在相同Key但是Value不相同時，是否覆蓋Value。
    Default: false
RamRole: ''
Tasks:
  - Name: getOnsInstance
    Action: ACS::ExecuteAPI
    Description: OnsInstanceInServiceList
    Properties:
      Service: ONS
      API: OnsInstanceInServiceList
      Parameters:
        RegionId: '{{ACS::TaskLoopItem}}'
    Outputs:
      onsInstance:
        Type: List
        ValueSelector: '.Data.InstanceVO[] |. + {"RegionId": "{{ACS::TaskLoopItem}}" }'
    Loop:
      RateControl:
        Mode: Concurrency
        MaxErrors: 100
        Concurrency: 1
      Items: '{{regionIds}}'
      Outputs:
        allOnsInstances:
          AggregateType: Fn::ListJoin
          AggregateField: onsInstance
  - Name: isTagTopic
    Action: ACS::Choice
    Description:
      en: Choose next task by resource type
      zh-cn: 根據資源類型選擇下一個任務
    Properties:
      DefaultTask: isTagGroup
      Choices:
        - When:
            Fn::Equals:
              - true
              - Fn::Jq:
                  - First
                  - contains(["topic"])
                  - '{{resourceTypes}}'
          NextTask: tagTopic
  - Name: tagTopic
    Action: ACS::Template
    Description:
      en: Update topic tags by ons instance tags
      zh-cn: 通過RocketMQ執行個體標籤更新Topic的標籤
    Properties:
      TemplateName: Ons_topicExtendInstanceTags
      Parameters:
        regionId:
          Fn::Jq:
            - First
            - .RegionId
            - '{{ACS::TaskLoopItem}}'
        instanceId:
          Fn::Jq:
            - First
            - .InstanceId
            - '{{ACS::TaskLoopItem}}'
        tagKeys:
          Fn::Jq:
            - All
            - .[].Key
            - '{{tags}}'
        isUpdate: '{{isUpdate}}'
    Loop:
      RateControl:
        Mode: Concurrency
        MaxErrors: 100
        Concurrency: 1
      Items:
        Fn::Jq:
          - All
          - .[][]
          - '{{getOnsInstance.allOnsInstances}}'
      Outputs:
        tagResult:
          AggregateType: Fn::ListJoin
          AggregateField: reqResult
  - Name: isTagGroup
    Action: ACS::Choice
    Description:
      en: Choose next task by resource type
      zh-cn: 根據資源類型選擇下一個任務
    Properties:
      DefaultTask: ACS::END
      Choices:
        - When:
            Fn::Equals:
              - true
              - Fn::Jq:
                  - First
                  - contains(["group"])
                  - '{{resourceTypes}}'
          NextTask: tagGroup
  - Name: tagGroup
    Action: ACS::Template
    Description:
      en: Update snapshot tags by ecs instance tags
      zh-cn: 通過RocketMQ執行個體標籤更新Group的標籤
    Properties:
      TemplateName: Ons_groupExtendInstanceTags
      Parameters:
        regionId:
          Fn::Jq:
            - First
            - .RegionId
            - '{{ACS::TaskLoopItem}}'
        instanceId:
          Fn::Jq:
            - First
            - .InstanceId
            - '{{ACS::TaskLoopItem}}'
        tagKeys:
          Fn::Jq:
            - All
            - .[].Key
            - '{{tags}}'
        isUpdate: '{{isUpdate}}'
    Loop:
      RateControl:
        Mode: Concurrency
        MaxErrors: 100
        Concurrency: 1
      Items:
        Fn::Jq:
          - All
          - .[][]
          - '{{getOnsInstance.allOnsInstances}}'
      Outputs:
        tagResult:
          AggregateType: Fn::ListJoin
          AggregateField: reqResult