rmq

Author: vladrbg

README.md

@@ -202,6 +202,31 @@ task mysql:query:source
 task mysql:query:branch SCENARIO=env-val
 ```
 
+### RabbitMQ Queue Splitting
+
+```bash
+# Fresh cluster with RabbitMQ from scratch
+task test:rmq:clean
+
+# Open the RabbitMQ management UI (http://localhost:15672)
+task rmq:management
+
+# Show RabbitMQ connection credentials
+task rmq:credentials
+
+# List mirrord temp queues left in the broker
+task rmq:queues:mirrord
+
+# Delete all mirrord temp queues from the broker
+task rmq:queues:mirrord:delete
+
+# Publish a test message with a tenant header
+task rmq:send QUEUE="orders" TENANT="a" MESSAGE="hello"
+
+# Run the e2e tests (from the operator/ directory)
+cargo test -p tests -- --ignored rmq_queue_splitting --nocapture
+```
+
 ### SQS/Kafka
 
 ```bash

Taskfile.yml

@@ -27,6 +27,9 @@ includes:
   multicluster:
     taskfile: ./tasks/Taskfile.multicluster.yml
     dir: ./tasks
+  rmq:
+    taskfile: ./tasks/Taskfile.rmq.yml
+    dir: ./tasks
   eks:
     taskfile: ./tasks/Taskfile.eks.yml
     dir: ./tasks
@@ -35,7 +38,7 @@ vars:
   ROOT_DIR:
     sh: pwd
   OPERATOR_DIR: '{{.OPERATOR_DIR | default (print .ROOT_DIR "/../operator")}}'
-  CHARTS_DIR: '{{.CHARTS_DIR | default (print .ROOT_DIR "/../charts/mirrord-operator")}}'
+  CHARTS_DIR: '{{.CHARTS_DIR | default (print .ROOT_DIR "/../operator/charts/mirrord-operator")}}'
   SCRIPTS_DIR: '{{.SCRIPTS_DIR | default (print .ROOT_DIR "/scripts")}}'
   MIRRORD_DIR: '{{.MIRRORD_DIR | default (print .ROOT_DIR "/../mirrord")}}'
   OPERATOR_IMAGE: '{{.OPERATOR_IMAGE | default "mirrord-operator:custom"}}'
@@ -128,6 +131,15 @@ tasks:
           operator
       - task: operator:image:load
 
+  build:mirrord:agent:
+    desc: Build mirrord agent image and load into minikube
+    vars:
+      AGENT_IMAGE: '{{.AGENT_IMAGE | default "test"}}'
+    cmds:
+      - cd {{.MIRRORD_DIR}} && {{.CONTAINER_RUNTIME}} build -f mirrord/agent/Dockerfile -t {{.AGENT_IMAGE}} .
+      - task: mirrord:agent:image:load
+        vars: { AGENT_IMAGE: "{{.AGENT_IMAGE}}" }
+
   build:mirrord:agent:gcs:
     desc: Build mirrord agent with GCS sccache
     vars:
@@ -687,6 +699,30 @@ tasks:
       - task: operator:install
       - task: sqs:deploy:aws
 
+  test:rmq:
+    desc: "Deploy RabbitMQ Cluster Operator + cluster and install operator"
+    deps: [operator:install]
+    cmds:
+      - task: rmq:deploy
+
+  test:rmq:clean:
+    desc: "Clean cluster and deploy RabbitMQ from scratch"
+    cmds:
+      - task: cluster:delete
+      - task: cluster:create
+      - task: build:mirrord:agent
+      - task: operator:install
+      - task: rmq:deploy
+
+  test:rmq:clean:nobuild:
+    desc: "Clean cluster and deploy RabbitMQ (skip operator build)"
+    cmds:
+      - task: cluster:delete
+      - task: cluster:create
+      - task: build:mirrord:agent
+      - task: operator:install:nobuild
+      - task: rmq:deploy
+
   test:kafka:
     desc: "Run Kafka test"
     deps: [build:app:kafka, operator:install]

k8s/overlays/multicluster-sqs/localstack-single.yaml

@@ -18,7 +18,7 @@ spec:
     spec:
       containers:
       - name: localstack
-        image: localstack/localstack:latest
+        image: localstack/localstack:4.14.0
         env:
         - name: SERVICES
           value: "sqs"

k8s/overlays/multicluster-sqs/localstack.yaml

@@ -15,7 +15,7 @@ spec:
     spec:
       containers:
       - name: localstack
-        image: localstack/localstack:latest
+        image: localstack/localstack:4.14.0
         env:
         - name: SERVICES
           value: "sqs"

k8s/overlays/sqs-localstack/localstack.yaml

@@ -15,7 +15,7 @@ spec:
     spec:
       containers:
       - name: localstack
-        image: localstack/localstack:latest
+        image: localstack/localstack:4.14.0
         env:
         - name: SERVICES
           value: "sqs"

operator-values.yaml

@@ -1,6 +1,7 @@
 operator:
   image: mirrord-operator
   sqsSplitting: true
+  rmqSplitting: true
   kafkaSplitting: true
   mysqlBranching: true
   pgBranching: true

tasks/Taskfile.multicluster.yml

@@ -1634,6 +1634,29 @@ tasks:
         fi
       - echo "All sessions cleaned up"
 
+  test:cleanup:resources:
+    desc: Delete leftover test deployments, services, and copy pods across all clusters
+    cmds:
+      - |
+        for ctx in {{.PRIMARY_CLUSTER}} {{.REMOTE_CLUSTER_1}}; do
+          echo "=== Cleaning $ctx ==="
+          kubectl --context $ctx delete deployments,services -n {{.MC_NAMESPACE}} -l mirrord-e2e-test-resource=true --ignore-not-found=true --wait=false
+          kubectl --context $ctx delete pods -n {{.MC_NAMESPACE}} -l operator.metalbear.co/resource-type=copy-target --ignore-not-found=true --wait=false
+        done
+      - |
+        if [ "{{.NUM_CLUSTERS}}" = "3" ]; then
+          echo "=== Cleaning {{.REMOTE_CLUSTER_2}} ==="
+          kubectl --context {{.REMOTE_CLUSTER_2}} delete deployments,services -n {{.MC_NAMESPACE}} -l mirrord-e2e-test-resource=true --ignore-not-found=true --wait=false
+          kubectl --context {{.REMOTE_CLUSTER_2}} delete pods -n {{.MC_NAMESPACE}} -l operator.metalbear.co/resource-type=copy-target --ignore-not-found=true --wait=false
+        fi
+      - echo "All leftover test resources cleaned up"
+
+  test:cleanup:all:
+    desc: Clean up all test sessions and leftover resources
+    cmds:
+      - task: test:cleanup:sessions
+      - task: test:cleanup:resources
+
   test:full:
     desc: Run full multi-cluster test setup
     cmds: