Ctrlk
HomeBlogSign In
slack
For the complete documentation index, see llms.txt. This page is also available as Markdown.
BetaTeamEnterprise

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:

A K8s application that consumes 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:

One SQS splitting session

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.

Two SQS splitting sessions

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

1

Enable SQS splitting in the Helm chart

Enable the operator.sqsSplitting setting in the mirrord-operator Helm chart.

2

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:

SQS Permission
needed for your queues
needed for temporary queues

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=true or env=dev etc. 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:Encrypt

  • kms:Decrypt

  • kms: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:GetObject on the relevant buckets / object prefixes.

  • s3:ListBucket on the relevant buckets if you want S3 to return 404 Not Found for deleted or missing objects instead of 403 Forbidden.

3

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.

4

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:

  1. It targets the deployment meme-app in namespace meme, restricted to container main.

  2. The container consumes two SQS queues. Their names are read from environment variables INCOMING_MEME_QUEUE_NAME and AD_QUEUE_NAME.

  3. The SQS queues can be referenced in a mirrord config under IDs meme-queue and ad-queue, respectively.

  4. The meme-queue has extra per-queue options defined in the meme-queue-options MirrordPropertyList (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 be sqs.

  • 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 with env). 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 (named value, 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 a MirrordPropertyList holding 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 to jq_filter as S3Metadata.

    • S3 to SQS (direct): set only s3_event: "true".

    • S3 to SNS to SQS: set both sns: "true" and s3_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:

  1. defined directly in the workload's pod template, with the value defined in value or in valueFrom via config map reference; or

  2. loaded 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 MirrordSplitConfig under ID meme-queue. All received messages will have an SQS attribute baggage containing mirrord-session=alice.

  • Will receive a subset of messages from SQS queues described in the MirrordSplitConfig under ID orders-queue. All received messages will have a JSON body with "important": true.

  • Will receive no messages from SQS queues described in the MirrordSplitConfig under ID ad-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-queue only when the message body is valid JSON and contains "client": "a".

  • Will receive messages from SQS queue fifo-orders-queue only when the SQS message attribute client has the value a.

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 client has the value a.

  • 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:

  1. Make sure a MirrordSplitConfig exists for the workload you're targeting:

  2. Note the queue-ids in the mirrord configuration have to match the queue-ids in the MirrordSplitConfig of the used target.

  3. 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.logLevel helm value to mirrord=info,operator=info,operator_sqs_splitting::forwarder=trace:

    or by setting the RUST_LOG environment variable in the operator's deployment to mirrord=info,operator=info,operator_sqs_splitting::forwarder=trace, e.g. by using kubectl edit deploy mirrord-operator -n mirrord.

  4. Some operations, like changing a MirrordSplitConfig of 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:

    1. Delete all the mirrord split session resources of the affected target. Those resources, MirrordClusterSplitSession, are not the same as MirrordSplitConfig. They are cluster-scoped, and reference the target by namespace and workload name. You can delete the ones for your target with:

    2. 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.

PreviousQueue SplittingNextKafka

Last updated

Was this helpful?