Amazon SQS
This page covers queue splitting for Amazon SQS. For the general concepts and the message filter reference shared by all queue services, see the Queue Splitting overview.
Queue splitting via MirrordSplitConfig requires mirrord operator 3.170.0 or later and mirrord CLI 3.221.0 or later.
⚠️ Deprecated CRD
MirrordWorkloadQueueRegistry is deprecated and replaced by MirrordSplitConfig. Existing resources continue to work for backward compatibility, but we recommend migrating to MirrordSplitConfig. See Migrating to MirrordSplitConfig.
The older operator.sqsSplittingLingerTimeout Helm value only affects legacy MirrordWorkloadQueueRegistry; with MirrordSplitConfig, use spec.drainTimeout instead.
How It Works
First, we have a consumer app reading messages from an SQS queue:
When the first mirrord SQS splitting session starts, two temporary queues are created (one for the target deployed in the cluster, one for the user's local application), and the mirrord operator routes messages according to the user's filter:
If a second user then starts a mirrord SQS splitting session on the same queue, a third temporary queue is created (for the second user's local application). The mirrord operator includes the new queue and the second user's filter in the routing logic.
If the filters defined by the two users both match some message, one of the users will receive the message at random.
Deployed targets will keep reading from the temporary queues as long as their temporary queues have unconsumed messages, so no messages intended for the remote service are lost.
Enabling SQS Splitting in Your Cluster
Enable SQS splitting in the Helm chart
Enable the operator.sqsSplitting setting in the mirrord-operator Helm chart.
Authenticate and authorize the mirrord operator
The mirrord operator will need to be able to perform operations on the SQS queues. To do this, it will build an SQS client, using the default credentials provider chain.
The easiest way to provide the credentials for the operator is with IAM role assumption. For that, an IAM role with an appropriate policy has to be assigned to the operator's service account. Please follow AWS's documentation on how to do that. Note that operator's service account can be annotated with the IAM role's ARN with the sa.roleArn setting in the mirrord-operator Helm chart.
Some of the SQS permissions are needed for your actual queues that you would like to split, and some permissions are only needed for the temporary queues, managed by the operator. Here is an overview:
GetQueueUrl
✓
ListQueueTags
✓
ReceiveMessage
✓
DeleteMessage
✓
GetQueueAttributes
✓
✓ (both!)
CreateQueue
✓
TagQueue
✓
SendMessage
✓
DeleteQueue
✓
And here we provide a short explanation for each required permission:
sqs:GetQueueUrl: the operator finds queue names to split in the provided source, and then it fetches the URL from SQS in order to make other API calls.sqs:GetQueueAttributes: the operator queries your queue's attributes, in order to clone these attributes to all derived temporary queues. It also reads the attributes of the temporary queues, in order to check the number of remaining messages.sqs:ListQueueTags: the operator queries your queue's tags, in order to clone these tags to all derived temporary queues.sqs:ReceiveMessage: the operator reads messages from the queues you split.sqs:DeleteMessage: after reading a message and forwarding it to a temporary queue, the operator deletes the message from the split queue.sqs:CreateQueue: the operator creates temporary queues in your SQS account.sqs:TagQueue: the operator sets tags on the temporary queues.sqs:SendMessage: the operator sends messages to the temporary queues.sqs:DeleteQueue: the operator deletes stale temporary queues in the background.
This is an example for a policy that gives the operator's roles the minimal permissions it needs to split a queue called ClientUploads:
The first statement gives the role the permissions it needs for your original queues.
Instead of specifying the queues you would like to be able to split in the first statement, you could alternatively make that statement apply for all resources in the account, and limit the queues it applies to using conditions instead of resource names. For example, you could add a condition that makes the statement only apply to queues with the tag
splittable=trueorenv=devetc. and set those tags for all queues you would like to allow the operator to split.The second statement in the example gives the role the permissions it needs for the temporary queues. Since all the temporary queues created by mirrord are created with the name prefix
mirrord-, that statement in the example is limited to resources with that prefix in their name.If you would like to limit the second statement with conditions instead of (only) with the resource name, you can set a condition that requires a tag, and for each queue you can specify tags that mirrord will set on the temporary queues it creates for that original queue (see per-queue options).
If the queue messages are encrypted, the operator's IAM role should also have the following permissions:
kms:Encryptkms:Decryptkms:GenerateDataKey
If you enable s3_event for an SQS queue (via its queueConfig), the operator will also fetch S3 object metadata for jq filtering. In that case, grant the operator:
s3:GetObjecton the relevant buckets / object prefixes.s3:ListBucketon the relevant buckets if you want S3 to return404 Not Foundfor deleted or missing objects instead of403 Forbidden.
Authorize deployed consumers
In order to be targeted with SQS splitting, a deployed consumer must be able to use the temporary queues created by mirrord. E.g. if the consumer application retrieves the queue's URL based on its name, lists queue's tags, consumes and deletes messages from the queue — it must be able to do the same on a temporary queue.
Any temporary queues managed by mirrord are created with the same policy as the original queues they are splitting (with the single change of updating the queue name in the policy). Therefore, access control based on SQS policies should automatically be taken care of.
However, if the consumer's access to the queue is controlled by an IAM policy (and not an SQS policy, see SQS docs), you will need to adjust it.
Provide application context
On operator installation with operator.sqsSplitting enabled, a new CustomResource type is defined in your cluster - MirrordSplitConfig. Users with permissions to get CRDs can verify its existence with kubectl get crd mirrordsplitconfigs.queues.mirrord.metalbear.co. Before you can run sessions with SQS splitting, you must create a MirrordSplitConfig for the desired target. This is because the MirrordSplitConfig contains additional application context required by the mirrord operator. For example, the operator needs to know which environment variables contain the names of the SQS queues to split.
See an example MirrordSplitConfig defined for a deployment meme-app living in namespace meme:
The MirrordSplitConfig above says that:
It targets the deployment
meme-appin namespacememe, restricted to containermain.The container consumes two SQS queues. Their names are read from environment variables
INCOMING_MEME_QUEUE_NAMEandAD_QUEUE_NAME.The SQS queues can be referenced in a mirrord config under IDs
meme-queueandad-queue, respectively.The
meme-queuehas extra per-queue options defined in thememe-queue-optionsMirrordPropertyList(e.g. tags set on temporary queues, SNS parsing).
Link the config to the deployed consumer
The MirrordSplitConfig is a namespaced resource, so it can only reference a consumer deployed in the same namespace. The target workload reference is specified with spec.targetRef:
apiVersion- API version of the Kubernetes workload (e.g.apps/v1).kind- type of the workload. mirrord supports SQS splitting on deployments and Argo rollouts.name- name of the workload.
Describe consumed queues
Each entry in the spec.queues list describes one or more SQS queues consumed by the workload:
id- arbitrary queue ID, as it will only be referenced from the user's mirrord config.kind- must besqs.appConfig.queue- how the application discovers the queue name/URL. Each entry can use:env- exact name of a single environment variable containing the queue name or URL.envLike- regex selecting multiple environment variables by name.fallback- fallback name/URL if the variable is not found (only valid withenv). The temporary queue name is still injected into the variable.valueSelector- a jq expression to extract queue names from the variable's value. Use.[]when the variable holds a JSON object whose string values are queue names/URLs.valuePattern- a regex used when the queue name/URL is embedded in a larger string. The capture group (namedvalue, otherwise the first group) marks the part that is the name; only that part is swapped for the temporary queue and the surrounding text is kept as-is.containers- limit resolution to specific containers (optional, defaults to all containers).
queueConfig(optional) - name of aMirrordPropertyListholding per-queue options (see below).
Per-queue options (SNS, S3 events, tags)
SQS-specific options live in a MirrordPropertyList referenced by the queue's queueConfig. The property list must be in the same namespace as the MirrordSplitConfig. Supported properties:
sns("true"/"false") - set to"true"if the queue contains SQS messages created from SNS notifications. Message bodies are then parsed and matched against users' filters, since SNS notification attributes live in the SQS message body. Defaults to"false".s3_event("true"/"false") - set to"true"to parse incoming messages as S3 event notifications and, on success, fetch user-defined S3 object metadata for the referenced object and expose it tojq_filterasS3Metadata.S3 to SQS (direct): set only
s3_event: "true".S3 to SNS to SQS: set both
sns: "true"ands3_event: "true". Defaults to"false".
tags- a JSON object string of tags that mirrord sets on every temporary queue it creates for this queue, e.g.'{"tool":"mirrord"}'.
For example, a MirrordPropertyList for a queue that tags its temporary queues and receives S3 events delivered through SNS:
The mirrord operator can only read consumer's environment variables if they are either:
defined directly in the workload's pod template, with the value defined in
valueor invalueFromvia config map reference; orloaded from config maps using
envFrom.
Setting a filter
For the full filter reference (queue_type, message_filter, jq_filter), see the overview. SQS uses queue_type: SQS.
Filtering on SQS message attributes, an unmatched (match-none) queue, and a jq_filter on the message body:
In the example above, the local application:
Will receive a subset of messages from SQS queues described in the
MirrordSplitConfigunder IDmeme-queue. All received messages will have an SQS attributebaggagecontainingmirrord-session=alice.Will receive a subset of messages from SQS queues described in the
MirrordSplitConfigunder IDorders-queue. All received messages will have a JSON body with"important": true.Will receive no messages from SQS queues described in the
MirrordSplitConfigunder IDad-queue.
A jq_filter on the message body or on a message attribute:
In the example above, the local application:
Will receive messages from SQS queue
orders-queueonly when the message body is valid JSON and contains"client": "a".Will receive messages from SQS queue
fifo-orders-queueonly when the SQS message attributeclienthas the valuea.
Filtering on S3 object metadata (requires s3_event: "true" on the queue's queueConfig):
In the example above, the local application will receive messages from SQS queue uploads-queue only when the S3 object referenced by the event has S3Metadata.client == "a".
Combining an attribute filter with a jq_filter (both must match):
In the example above, the local application will receive messages from SQS queue orders-queue only when both of these conditions hold:
The SQS message attribute
clienthas the valuea.The JSON message body contains
"important": true.
Using the * wildcard to apply one filter to all queues described in the MirrordSplitConfig:
In the example above, the local application will receive a subset of message from all SQS queues described in the MirrordSplitConfig. All received messages will have an SQS attribute baggage containing mirrord-session=pr-123. * is a special queue ID for SQS queues, and resolves to all queues described in the MirrordSplitConfig.
Troubleshooting SQS splitting
If you're trying to use SQS-splitting and are facing difficulties, here are some steps you can go through to identify and hopefully solve the problem.
First, some generally applicable steps:
Make sure a
MirrordSplitConfigexists for the workload you're targeting:Note the queue-ids in the mirrord configuration have to match the queue-ids in the
MirrordSplitConfigof the used target.Get the logs from the mirrord-operator, in case it becomes necessary for the mirrord team to look into your issue, like this:
If that fails, you might not have permissions to the operator's logs.
To get especially helpful logs, you can change the log level for SQS-splitting, then reproduce the issue to get the relevant logs. This can be achieved by reinstalling the helm chart and setting the
operator.logLevelhelm value tomirrord=info,operator=info,operator_sqs_splitting::forwarder=trace:or by setting the
RUST_LOGenvironment variable in the operator's deployment tomirrord=info,operator=info,operator_sqs_splitting::forwarder=trace, e.g. by usingkubectl edit deploy mirrord-operator -n mirrord.Some operations, like changing a
MirrordSplitConfigof a workload while there are active sessions to that target, are not yet supported, and can lead to a bad state for mirrord. If you've reached such a state:Delete all the mirrord split session resources of the affected target. Those resources,
MirrordClusterSplitSession, are not the same asMirrordSplitConfig. They are cluster-scoped, and reference the target by namespace and workload name. You can delete the ones for your target with:Restart the mirrord operator, e.g.:
If some (but not all) of the messages that should arrive at the local service arrive at the remote service
It's possible the target workload's restart is not complete yet, and there are still pods reading directly from the original queue (those will be pods that DO NOT have a operator.metalbear.co/patched label). You can wait a bit for them to be replaced with new pods, patched by mirrord, that read from a temporary queue created by mirrord, or you can delete them.
If all SQS sessions are over but the remote service still didn't change back to read from the original queue
When there are no more queue splitting sessions to a target, the target workload will not immediately be changed to read directly from the original queue. Instead, it will keep reading from the temporary queue until its empty, so that no messages intended for the remote service are lost.
If the target workload doesn't change back within the expected time, check its logs and make sure it is consuming queue messages.
If you don't want to wait for the remote service to drain the temporary queue, and you don't care about losing those messages, you can cap the wait with spec.drainTimeout (in seconds) on the MirrordSplitConfig:
spec.drainTimeout
Behavior
unset
Drain indefinitely - the temporary queue is kept until the remote service empties it.
0
Skip draining; delete the temporary queue immediately. Unread messages may be lost.
N
Wait up to N seconds for the queue to drain, then delete it.
If that service is trying to consume messages correctly, and the temporary queue is already empty, but the target application still doesn't get restored to its original state, please try restarting the application, deleting any lingering MirrordClusterSplitSession objects, and if possible, restart the mirrord operator.
Last updated
Was this helpful?

