Asos.ServiceBus.MessageSiphon 1.0.20-ci-message-selec0003

azure-servicebus-message-siphon

A dotnet tool for message siphon scenarios when working with Azure Service Bus, allowing you to purge and replay messages, within a single namespace or across namespaces

What's it for?

You may have cases such as

• Messages have dead lettered and you want to replay them back onto the main queue\topic
• You want to clone messages from one entity to another in the same namespace
• You want to clone and remove messages from one namespace to another
• You want to select message by a header value and move them
• You want to purge messages, deleting from a queue, topic or dead-letter sub entity.
• You want to use connection strings, RBAC or both for the source and targets. (i.e. connection string for source, RBAC for target)

How does it work?

A configuration file is used that determines the set of work that will be performed. The configuration file is made up of service bus connection details, and a list of siphon work that needs to be performed. The siphon work then references the connections by Name

As the consumer of this package, you can define some standard work you'd like to perform as a set of configuration files, then use them to execute the work against one or many service bus namespaces - just provide the appropriate ConnectionString or FullyQualifiedNamespace at runtime

Configuration details

ReplayMessagesJob - A replay messages just has the following

• JobType - the type of work to perform. Can be one of DeadLettersSameEntity, DeadLettersToDifferentEntity or SourceToTarget
  • DeadLettersSameEntity - replays messages from the sub-queue back onto the parent entity
  • DeadLettersToDifferentEntity - replays messages from the entity sub-queue, to a different Target entity
  • SourceToTarget - siphons messages from any source to any target entity
  • PurgeSource - purges messages from the source entity, a delete operation
  • PurgeDeadLetters - as per PurgeSource but for dead letter entities
• JobName - Give the work a useful name, used in log messages etc
• NumberOfConcurrentProcesses - When performing a siphon job, copying messages from source to target, messages are received in batches as per the SourceBatchReceiveSize setting. Once messages are received, they need to be sent to the target and NumberOfConcurrentProcesses controls how many concurrent operations attempt to work through the batch. E.g. if your batch size is 100 and you set NumberOfConcurrentProcesses to 10, then 10 operations will each attempt to process 10 messages in parallel. Experiment with these settings to find the best combination to achieve high throughput.

ServiceBusDetails - a list of service bus connections. A ServiceBusDetail instance has the following

• Name - give the item a useful name, you'll use this to refer to it from any siphon work you define
• ConnectionMode - how to connect to the service bus, can be one of ConnectionString or Rbac
• ConnectionString - if ConnectionMode is ConnectionString, then define the SAS connection string to use
• FullyQualifiedNamespace = if ConnectionMode is Rbac, then define the fully qualified namespace, e.g. namespace.servicebus.windows.net

SiphonWork - a list of work to perform, Siphon work requires the following

• SiphonMode - the type of operation to perform. Can be Clone or CloneAndDelete.

  • When Clone, the worker will use PeekMessagesAsync so not to increment the delivery count of messages, and the message will not be received or completed. A copy of the message is created
  • When CloneAndDelete, the worker will use ReceiveMessagesAsync, and will attempt to complete the message once the send operation to target has completed OK

• SourceConnectionName - the Name of the ServiceBusDetails to use as the source

• SourceEntity - the name of the entity, can be either a queue or a topic

• SourceSubscriptions - if SourceEntity is a topic, then define a list of subscription names for the Topic

• TargetConnectionName - the Name of the ServiceBusDetails to use as the target

• TargetEntity - the name of the entity to send to, can be either a queue or a topic

• SourceBatchReceiveSize - the size of the batch to recieve messages from the source. Defaults to 20.

• MessagesOlderThan - An ISO8601 date span, as per the duration specification. E.g. PT72H (72 hours), P30D (30 days). When performing a purge, only messages that have an older enqueue time than this value are considered, allowing you to perform work such as purge messages older than 72 hours

• HeaderSelector - A filter to use on the message headers to select only matching messages. This has the following limitations

  • Only a single header filter is support - You can filter by one field only, e.g. HeaderName = Value
  • Only =, > and < operators are supported
  • If you use > or < operators, then the field selector must be numeric, e.g. HeaderName > 12345

Example configurations

Replaying service bus messages from dead letters on the subscriptions

In this configuration, we only need to define a single service bus connection, since we're replaying onto the same namespace/entity.

{
    "Logging": {
        "LogLevel": {
            "Default": "Information"
        }
    },    
    "ReplayMessagesJob": {
        "JobType": "DeadLettersSameEntity",
        "JobName": "Replay-Messages-From-DeadLetter-Subqueue",
        "NumberOfConcurrentProcesses": 1,
        "ServiceBusDetails": [
            {
                "Name": "Source",
                "ConnectionMode": "ConnectionString",
                "ConnectionString": "Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=your-sas-key"
            }
        ],
        "SiphonWork": [
            {
                "SiphonMode": "CloneAndDelete",
                "SourceConnectionName": "Source",
                "SourceEntity": "snapshot",
                "SourceSubscriptions": [ "test", "test1" ]
            }
        ]
    }
}

Cloning messages from a subscription in one namespace using a connection string, to another namespace using RBAC.

In this example, messages will be receieved in batches of 40, and 5 concurrent tasks will be used to to send them to the target in parallel (i.e. each task processes 8 messages each)

{
    "Logging": {
        "LogLevel": {
        }
    },    
    "ReplayMessagesJob": {
        "JobType": "SourceToTarget",
        "JobName": "Local Debug",
        "NumberOfConcurrentProcesses": 5,
        "ServiceBusDetails": [
            {
                "Name": "Source",
                "ConnectionMode": "ConnectionString",
                "ConnectionString": "Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=access-key"
            },
            {
                "Name": "Target",
                "ConnectionMode": "Rbac",
                "FullyQualifiedNamespace": "namespace1.servicebus.windows.net"
            }
        ],
        "SiphonWork": [
            {
                "SiphonMode": "Clone",
                "SourceConnectionName": "Source",
                "SourceEntity": "topic-entity",
                "SourceSubscriptions": [ "test-subscription" ],
		"SourceBatchReceiveSize" : 40,
                "TargetConnectionName": "Target",
                "TargetEntity": "copy-of-topic"
            }
        ]
    }
}

Cloning messages from a subscription in one namespace to another topic in the same namespace, using a message selector to filter .

In this example, only messages that has a header

{
    "Logging": {
        "LogLevel": {
        }
    },    
    "ReplayMessagesJob": {
        "JobType": "SourceToTargetByMessageSelector",
        "JobName": "Local Debug",
        "ServiceBusDetails": [
            {
                "Name": "Source",
                "ConnectionMode": "ConnectionString",
                "ConnectionString": "Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=access-key"
            }
        ],
        "SiphonWork": [
            {
                "SiphonMode": "Clone",
                "SourceConnectionName": "Source",
                "SourceEntity": "topic-entity",
                "SourceSubscriptions": [ "test-subscription" ],
                "HeaderSelector": "[Payload.MessageType] = 'ProductUpdated'", 
                "SourceBatchReceiveSize" : 40,
                "TargetConnectionName": "Source",
                "TargetEntity": "copy-of-topic"
            }
        ]
    }
}

Using RBAC, receive messages from a dead letter subscription in batches of 100 and purge messages that are older than 1 hour

{
    "Logging": {
        "LogLevel": {
            "Default": "Information"
        }
    },   
    "ReplayMessagesJob": {
        "JobType": "PurgeDeadLetters",
        "JobName": "Purge Debug",
        "ServiceBusDetails": [
            {
                "Name": "Source",
                "ConnectionMode": "Rbac",
                "FullyQualifiedNamespace": "namespace.servicebus.windows.net"
            }
        ],
        "SiphonWork": [
            {
                "SiphonMode": "Delete",
                "SourceConnectionName": "Source",
                "SourceEntity": "demo",
                "SourceSubscriptions": [ "test" ],
                "SourceBatchReceiveSize" : 100,
                "MessagesOlderThan" : "PT1H"
            }
        ]
    }
}

How to use?

The package is available as a dotnet tool, so you can install from nuget.org by running the following command

dotnet tool update --global Asos.ServiceBus.MessageSiphon ---version x.x.x

Once installed, you have a command available to you siphon-asb-messages. This requires a single parameter, the path to the configuration file, e.g.

siphon-asb-messages -n D:\temp\replay-config.json

siphon-asb-messages -n usr/tmp/replay-config.json

Azure Pipelines

Since this is a dotnet tool, you can run it from your Azure Pipeline. This lets you take advantage of Managed Identity from Service Connections, and interact with namespaces that may be IP Restricted by running the work from a build pool that is allow-listed on the namespace.

This gives you a way to allow supporting team members to run jobs, replaying or purging messages, in a secure and repeatable manner & without requring any particular tools or permissions.

For example, another team may have a dependency on a service bus that you own and maintain. The namespace is IP rescricted and access is via RBAC, they need to replay some dead letters on a topic-subscription. You can define an Azure pipeline to do this work and define a particular build pool with a known IP, then grant their team members permissions to run the pipeline. In this way, the consuming team can self-serve without sharing connection strings or having to use Bastion services

To use this, you'd need to define a configuration file in your source repository. Note, we use a variable placeholder here for $ServiceBusConnectionString

{
  "Logging": {
    "LogLevel": {
      "Default": "Information"
    }
  },  
  "ReplayMessagesJob": {
    "JobType": "DeadLettersSameEntity",
    "JobName": "Dead Letters Replay",
    "ServiceBusDetails": [
      {
        "Name": "Source",
        "ConnectionMode": "ConnectionString",
        "FullyQualifiedNamespace": "$(ServiceBusConnectionString)"
      }
    ],
    "SiphonWork": [
      {
        "SiphonMode": "CloneAndDelete",
        "SourceConnectionName": "Source",
        "SourceEntity": "some-topic-orders",
        "SourceSubscriptions": [
          "the-subscription-name"
        ]
      }
    ]
  }
}

An example pipeline might look like so. In this example, we support either RBAC using a fully qualified namespace, or connection strings by retrieving a secret from a vault. Replace Tokens will update the config file with the appropriate values

parameters:
  - name: StageName
    displayName: Service Bus Connection String Secret
    type: string
    default: "ReplayPurgeMessages"

  - name: AzureSubscription
    displayName: The name of the service connection that's used to connect to Azure
    type: string

  - name: ConfigPath
    displayName: Path to config
    type: string

  - name: ConfigFileName
    displayName: Message replay config
    type: string

  - name: Environment
    displayName: Environment
    type: string

  - name: ServiceBusFullyQualifiedName
    displayName: ServiceBusFullyQualifiedName
    type: string
    default: ""
  
  - name: KeyVaultSubscriptionName
    displayName: Name of the subscription that contains the key vault, if using it to retrieve connection strings
    type: string
    default: ""
  
  - name: KeyVaultName
    displayName: Key Vault Name
    type: string
    default: ""
  
  - name: SecretName
    displayName: Service Bus Connection String Secret
    type: string
    default: ""

stages:
  - stage: ${{parameters.StageName}}
    displayName: ${{parameters.StageName}}

    pool:
      vmImage: ubuntu-latest

    jobs: 
      - job:
        steps:
          - checkout: self

          - task: AzureCLI@2
            displayName: Get Service Bus Connection String From Vault
            condition: and(succeeded(), ne('${{ parameters.KeyVaultName }}', ''))
            inputs:
              azureSubscription: ${{parameters.AzureSubscription}}
              scriptType: pscore
              scriptLocation: inlineScript
              inlineScript: |
                $subscription = ${{parameters.KeyVaultSubscriptionName}}
                $connectionString = az keyvault secret show --name ${{parameters.SecretName}} --subscription $subscription --vault-name ${{parameters.KeyVaultName}} --query value -o tsv                
                echo "##vso[task.setvariable variable=ServiceBusConnectionString]$connectionString"

          - task: AzureCLI@2
            displayName: Set variable to fully qualified namespace parameter
            condition: and(succeeded(), eq('${{ parameters.KeyVaultName }}', ''))
            inputs:
              azureSubscription: ${{parameters.AzureSubscription}}
              scriptType: pscore
              scriptLocation: inlineScript
              inlineScript: |
                  $namespaceName = ${{parameters.ServiceBusFullyQualifiedName}}
                  echo "Connecting to namespace " + $namespaceName
                  echo "##vso[task.setvariable variable=ServiceBusFullyQualifiedName]$namespaceName"

          - task: replacetokens@4
            displayName: Replace Tokens
            inputs:
              targetFiles: '${{parameters.ConfigPath}}/*'
              encoding: 'auto'
              tokenPattern: 'azpipelines'
              writeBOM: true
              actionOnMissing: 'warn'
              keepToken: false
              actionOnNoFiles: 'continue'
              enableTransforms: false
              useLegacyPattern: false
              enableTelemetry: true

          - task: AzureCLI@2
            displayName: Perform Message Work
            inputs:
              azureSubscription: ${{parameters.AzureSubscription}}
              scriptType: pscore
              scriptLocation: inlineScript
              inlineScript: |
                dotnet tool update Asos.ServiceBus.MessageSiphon --tool-path . --version 1.0.0
                & "./siphon-asb-messages" -n ${{parameters.ConfigPath}}/${{parameters.ConfigFileName}}