> For the complete documentation index, see [llms.txt](https://metalbear.com/mirrord/docs/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://metalbear.com/mirrord/docs/sharing-the-cluster/queue-splitting/nats.md). # NATS JetStream This page covers queue splitting for [NATS JetStream](https://docs.nats.io/nats-concepts/jetstream). For the general concepts and the message filter reference shared by all queue services, see the [Queue Splitting overview](/mirrord/docs/sharing-the-cluster/queue-splitting.md). The word "queue" on this page refers to a JetStream stream consumed through a durable pull consumer. {% hint style="info" %} NATS splitting works with JetStream only: the application must consume through a durable pull consumer on a stream. Core NATS (plain subject subscriptions) is not supported yet. The NATS server must be version `2.2` or later, since splitting relies on message headers. {% endhint %} ### How It Works First, we have a consumer app reading messages from a JetStream stream through a durable pull consumer. When the first mirrord NATS splitting session starts, the operator creates a temporary main-output stream and a durable consumer on it - both deep copies of the originals' settings - and patches the workload's stream and consumer environment variables to point at them. The operator then reads messages through the application's original durable consumer and republishes each one to `.`: * A message that matches a user's filter is republished to that session's temporary stream, and the local application reads it from there. * A message that matches no filter is republished to the main-output stream, and the deployed workload reads it from there. ![NATS queue splitting flow](https://2838204941-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FghNlSpMkqkYKZCZQsHRt%2Fuploads%2Fgit-blob-8f313e1d1785505f759b9e015db037c31f26b9b5%2Fnats.svg?alt=media) In the default `steal` mode a matched message goes only to the session's temporary stream. In `mirror` mode it goes to both the session's temporary stream and the main output, so the deployed workload also processes a copy. Delivery is at-least-once: a message is acknowledged on the original consumer only after the JetStream server has acknowledged the republish, so a message can be redelivered but never silently dropped. When sessions end, their temporary streams are deleted, and when the whole split ends the main-output stream is deleted too. Cleanup is crash-safe: every temporary resource is tracked in a `MirrordClusterExternalResource`, so it is removed even if the operator restarts mid-session. ### Enabling NATS Splitting in Your Cluster {% stepper %} {% step %} **Enable NATS splitting in the Helm chart** Enable the `operator.natsSplitting` setting in the [mirrord-operator Helm chart](https://github.com/metalbear-co/charts/blob/main/mirrord-operator/values.yaml). {% endstep %} {% step %} **Create a MirrordPropertyList** The operator needs to connect to your NATS server to read and republish messages. Define the connection in a `MirrordPropertyList` ([`CustomResource`](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/)) in the same namespace as the target workload (and the `MirrordSplitConfig`), or in the operator's namespace to share it across namespaces - see [Sharing Property Lists Across Namespaces](/mirrord/docs/sharing-the-cluster/queue-splitting.md#sharing-property-lists-across-namespaces). ```yaml apiVersion: mirrord.metalbear.co/v1 kind: MirrordPropertyList metadata: name: nats-config namespace: events spec: properties: - name: url value: nats://nats.nats-ns.svc.cluster.local:4222 ``` Supported properties: | Property | Description | Required | Default | | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------ | :------: | :-----: | | `url` | NATS server URL, e.g. `nats://host:4222`. For a NATS cluster, a comma-separated list of URLs. Use the `tls://` scheme to connect over TLS. | Yes | | | `tls` | Set to `"true"` to connect over TLS without changing the URL scheme. | No | `false` | | `username` | Username, used together with `password`. | No | | | `password` | Password, used together with `username`. | No | | | `token` | Authentication token. | No | | | `credentials` | Contents of a JWT `.creds` file. | No | | | `nkey` | NKey seed. | No | | | `ca_cert` | PEM contents of the CA bundle that signs the server's certificate, for servers with a private CA. Implies TLS. | No | | | `client_cert` | PEM contents of the client certificate chain, for mTLS. Used together with `client_key`. Implies TLS. | No | | | `client_key` | PEM contents of the client certificate's private key. Used together with `client_cert`. | No | | Set the auth properties matching how your NATS server authenticates clients. Keep secrets in a Kubernetes Secret and reference them with `valueFrom`. Username and password: ```yaml spec: properties: - name: url value: nats://nats.nats-ns.svc.cluster.local:4222 - name: username value: mirrord-operator - name: password valueFrom: secretKeyRef: name: nats-auth key: password ``` A token: ```yaml - name: token valueFrom: secretKeyRef: name: nats-auth key: token ``` A JWT credentials file (the property holds the contents of the `.creds` file, not a path): ```yaml - name: credentials valueFrom: secretKeyRef: name: nats-auth key: nats.creds ``` An NKey seed: ```yaml - name: nkey valueFrom: secretKeyRef: name: nats-auth key: nkey.seed ``` TLS with a private CA, or mutual TLS (the properties hold the PEM contents, not paths; providing any of them makes the operator connect over TLS even on a `nats://` URL): ```yaml - name: ca_cert valueFrom: secretKeyRef: name: nats-tls key: ca.crt - name: client_cert valueFrom: secretKeyRef: name: nats-tls key: tls.crt - name: client_key valueFrom: secretKeyRef: name: nats-tls key: tls.key ``` Without `ca_cert`, the server's certificate is verified against the standard public roots, so `client_cert` + `client_key` alone give mTLS against a publicly-trusted server. {% endstep %} {% step %} **Create a MirrordSplitConfig** On operator installation with `operator.natsSplitting` enabled, a new [`CustomResource`](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/) 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`. Create a `MirrordSplitConfig` for the target workload. NATS uses `kind: nats` in queue entries. ```yaml apiVersion: queues.mirrord.metalbear.co/v1 kind: MirrordSplitConfig metadata: name: order-processor-split namespace: events spec: targetRef: apiVersion: apps/v1 kind: Deployment name: order-processor clientConfigs: nats: nats-config queues: - id: orders kind: nats appConfig: stream: - env: NATS_STREAM consumer: - env: NATS_CONSUMER ``` The `MirrordSplitConfig` above says that: 1. It targets the deployment `order-processor` in namespace `events`. 2. The NATS connection comes from the `nats-config` `MirrordPropertyList`. 3. The deployment consumes one stream, whose name is in environment variable `NATS_STREAM`, through a durable consumer whose name is in environment variable `NATS_CONSUMER`. 4. The stream can be referenced in a mirrord config under ID `orders`. **Link the config to the deployed consumer** The `MirrordSplitConfig` is a namespaced resource. 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. Supported: `Deployment`, `StatefulSet`, `Rollout`. * `name` - name of the workload. **Describe consumed streams** Each entry in the `spec.queues` list describes one stream and the durable pull consumer the workload reads it through: * `id` - arbitrary queue ID that developers [reference](#setting-a-filter) from their mirrord config. * `kind` - must be `nats`. * `clientConfig` (optional) - name of a `MirrordPropertyList` with the NATS connection. Can also be set once for all NATS queues with `spec.clientConfigs.nats`. * `queueConfig` (optional) - name of a `MirrordPropertyList` with settings for the temporary stream and consumer, see [Configuring temporary streams](#configuring-temporary-streams). * `appConfig.stream` - how the application discovers the stream name. Each entry can use the same structure as other queue services: `env`, `envLike`, `volume`, `fallback`, `valuePattern`, `containers`. * `appConfig.consumer` - how the application discovers the durable consumer name. Uses the same structure as `stream`. Each queue entry must resolve to exactly one stream and one consumer. If the workload runs several consumers, add one queue entry per consumer. {% hint style="warning" %} 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`. {% endhint %} {% endstep %} {% endstepper %} ### Configuring temporary streams By default the temporary streams and consumers mirrord creates are deep copies of the source stream and consumer, so they inherit their settings - including retention, acknowledgement wait, and replication. You can override these per queue by pointing its `queueConfig` at a `MirrordPropertyList`: ```yaml apiVersion: mirrord.metalbear.co/v1 kind: MirrordPropertyList metadata: name: orders-queue-config namespace: events spec: properties: - name: max_age_seconds value: "3600" - name: ack_wait_seconds value: "120" - name: num_replicas value: "1" ``` Reference it from the queue entry in the `MirrordSplitConfig`: ```yaml queues: - id: orders kind: nats queueConfig: orders-queue-config appConfig: stream: - env: NATS_STREAM consumer: - env: NATS_CONSUMER ``` * `max_age_seconds` (integer seconds) - how long the temporary stream keeps messages. * `ack_wait_seconds` (integer seconds) - how long the temporary consumer waits for an acknowledgement before redelivering a message. Raising it gives your local application more time to handle a message, which is useful when you pause on a breakpoint while debugging. * `num_replicas` (integer) - how many replicas the temporary stream has. Each key is optional, and any key you leave out keeps the value copied from the source stream or consumer. An invalid value for a key is ignored (a warning is logged) and that setting falls back to the source's value, so a typo never fails the session. ### Setting a filter For the full filter reference (`queue_type`, `message_filter`, `jq_filter`), see the [overview](/mirrord/docs/sharing-the-cluster/queue-splitting.md#setting-a-filter-for-a-mirrord-run). NATS uses `queue_type: NATS`. `message_filter` maps NATS message header names to regexes for their values. A message matches only when **all** entries match, and header name matching is case-sensitive. Filtering on a message header: ```json { "operator": true, "target": "deployment/order-processor/container/consumer", "feature": { "split_queues": { "orders": { "queue_type": "NATS", "message_filter": { "x-tenant": "^test$" } } } } } ``` In the example above, the local application will receive only messages carrying an `x-tenant` header with the value `test`. `jq_filter` runs a jq program on a JSON object with `subject`, `headers`, and `payload` fields. `payload` is the parsed message body when the body is valid JSON, and a plain string otherwise. Filtering on the message body with `jq_filter`: ```json { "operator": true, "target": "deployment/order-processor/container/consumer", "feature": { "split_queues": { "orders": { "queue_type": "NATS", "jq_filter": ".payload.user_id == \"test-user\"" } } } } ``` In the example above, the local application will receive only messages whose JSON body contains `"user_id": "test-user"`. {% hint style="info" %} When the operator's `operator.injectSessionKeyHeader` setting is enabled, every message delivered to your session carries a `mirrord-key` header with your session key - see [Session Key Header](/mirrord/docs/sharing-the-cluster/queue-splitting.md#session-key-header). {% endhint %} ### Notes and limitations * JetStream only. The application must consume through a durable pull consumer on a stream. Core NATS (plain subject subscriptions) is not supported yet. * The NATS server must be version `2.2` or later, since splitting relies on message headers. * Each queue entry in the `MirrordSplitConfig` describes exactly one stream and one consumer. Add one entry per consumer. * Republished messages carry the subject `.` - the original subject is kept, prefixed with the temporary stream's name. An application that routes on exact subjects sees the prefixed subject while a split is active, so match on the subject's suffix (or a wildcard) instead of the full subject. * Delivery is at-least-once: a message is acknowledged on the original consumer only after the JetStream server acknowledged the republish, so your application may see a message twice but never miss one. --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://metalbear.com/mirrord/docs/sharing-the-cluster/queue-splitting/nats.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.