Test an elasticache celery broker

environments/staging/inventory.ini

@@ -123,6 +123,9 @@ kafka17
 celery_a1
 celery_a2
 
+[celery_a2:vars]
+rabbitmq_migration_bridge=true
+
 [celery:vars]
 swap_size=2G
 

environments/staging/inventory.ini.j2

@@ -99,6 +99,9 @@ kafka17
 celery_a1
 celery_a2
 
+[celery_a2:vars]
+rabbitmq_migration_bridge=true
+
 [celery:vars]
 swap_size=2G
 

environments/staging/public.yml

@@ -21,8 +21,14 @@ TWO_FACTOR_GATEWAY_ENABLED: True
 erlang: 1:24.3.3-1
 rabbitmq_version: 3.10.13-1
 
+# Elasticache Broker Info
+ELASTICACHE_BROKER_DB: '0'
+ELASTICACHE_BROKER_HOST: "broker.staging.commcare.local"
+ELASTICACHE_BROKER_PORT: '6379'
+
 #RabbitMQ Broker URL
-BROKER_URL: "amqp://{{ AMQP_USER }}:{{ AMQP_PASSWORD }}@{{ AMQP_HOST }}:5672/{{ AMQP_NAME }};amqp://{{ AMQP_USER }}:{{ AMQP_PASSWORD }}@{{ groups.rabbitmq.1 }}:5672/{{ AMQP_NAME }}"
+BROKER_URL: "redis://{{ ELASTICACHE_BROKER_HOST }}:{{ ELASTICACHE_BROKER_PORT }}/{{ ELASTICACHE_BROKER_DB }}"
+OLD_BROKER_URL: "amqp://{{ AMQP_USER }}:{{ AMQP_PASSWORD }}@{{ AMQP_HOST }}:5672/{{ AMQP_NAME }};amqp://{{ AMQP_USER }}:{{ AMQP_PASSWORD }}@{{ groups.rabbitmq.1 }}:5672/{{ AMQP_NAME }}"
 
 backup_blobdb: False
 backup_postgres: plain
@@ -125,6 +131,7 @@ localsettings:
   BANK_SWIFT_CODE: 'CTZIUS33'
   CELERY_PERIODIC_QUEUE: 'celery_null'
   CELERY_REPEAT_RECORD_DATASOURCE_QUEUE: "repeat_record_datasource_queue"
+  CELERY_BROKER_POOL_LIMIT: 1
   COMMCARE_ANALYTICS_HOST: "https://commcare-analytics-staging.dimagi.com"
   COUCH_CACHE_DOCS: True
   COUCH_CACHE_VIEWS: True

environments/staging/terraform.yml

@@ -406,6 +406,22 @@ elasticache_cluster:
   snapshot_retention: 7
   snapshot_window: "07:30-08:30"
 
+elasticache_celery_broker:
+  create: yes
+  cache_engine: "redis"
+  cache_engine_version: "7.0"
+  cache_node_type: "cache.r6g.large"
+  params:
+    maxmemory-policy: "noeviction"
+  automatic_failover: true
+  transit_encryption: false
+  at_rest_encryption: true
+  auto_minor_version: false
+  cluster_size: 2
+  multi_az: true
+  maintenance_window: "sun:00:30-sun:01:30"
+  snapshot_retention: 0
+
 r53_private_zone:
   create: yes
   domain_name:  "staging.commcare.local"

src/commcare_cloud/ansible/roles/commcarehq/templates/localsettings.py.j2

@@ -391,6 +391,9 @@ CELERY_HEARTBEAT_THRESHOLDS = {
     {{ queue | to_json }}: {% if threshold == None %}None{% else %}{{ threshold | to_json }}{% endif %},
 {% endfor %}
 }
+{% if 'CELERY_BROKER_POOL_LIMIT' in localsettings and localsettings.CELERY_BROKER_POOL_LIMIT %}
+CELERY_BROKER_POOL_LIMIT = {{ localsettings.CELERY_BROKER_POOL_LIMIT }}
+{% endif %}
 
 {% if 'CELERY_REMINDER_CASE_UPDATE_BULK_QUEUE' in localsettings and localsettings.CELERY_REMINDER_CASE_UPDATE_BULK_QUEUE %}
 CELERY_REMINDER_CASE_UPDATE_BULK_QUEUE = '{{ localsettings.CELERY_REMINDER_CASE_UPDATE_BULK_QUEUE }}'

src/commcare_cloud/ansible/roles/commcarehq/templates/supervisor_celery_flower.conf.j2

@@ -5,9 +5,6 @@ command={{ virtualenv_home }}/bin/celery
     -A corehq flower
     --address=0.0.0.0
     --port={{ app_processes_config.flower_port }}
-    {% if 'rabbitmq' in groups and groups.rabbitmq %}
-    --broker_api=http://{{ AMQP_USER }}:{{ AMQP_PASSWORD }}@{{ groups.rabbitmq.0 }}:15672/api/
-    {% endif %}
 directory={{ code_home }}
 user={{ cchq_user }}
 numprocs=1

src/commcare_cloud/commands/terraform/templates/commcarehq.tf.j2

@@ -234,6 +234,32 @@ module "elasticache-module-base" {
 }
 {% endif %}
 
+{% if elasticache_celery_broker %}
+module "elasticache-celery-broker" {
+  source                = "./modules/elasticache-cluster"
+  create               = "{{ elasticache_celery_broker.create|tojson }}"
+  namespace             = "${var.environment}-elasticache-celery-broker"
+  cluster_id            = "${var.environment}-elasticache-celery-broker"
+  subnet_ids_cache      = values(module.network.subnets-app-private)
+  securitygroup_id      = compact([module.network.db-private-sg, module.network.ssh-sg, module.network.vpn-connections-sg])
+  cache_engine          = {{ elasticache_celery_broker.cache_engine|tojson }}
+  cache_engine_version  = {{ elasticache_celery_broker.cache_engine_version|tojson }}
+  cache_node_type       = {{ elasticache_celery_broker.cache_node_type|tojson }}
+  params                = {{ elasticache_celery_broker.params|tojson }}
+  automatic_failover    = {{ elasticache_celery_broker.automatic_failover|tojson }}
+  transit_encryption    = {{ elasticache_celery_broker.transit_encryption|tojson }}
+  at_rest_encryption    = {{ elasticache_celery_broker.at_rest_encryption|tojson }}
+  multi_az              = {{ elasticache_celery_broker.multi_az|tojson }}
+  auto_minor_version    = {{ elasticache_celery_broker.auto_minor_version|tojson }}
+  maintenance_window    = {{ elasticache_celery_broker.maintenance_window|tojson }}
+  snapshot_retention    = {{ elasticache_celery_broker.snapshot_retention|tojson }}
+  snapshot_window       = {{ elasticache_celery_broker.snapshot_window|tojson }}
+  port_number           = 6379
+  cluster_size          = {{ elasticache_celery_broker.cluster_size|tojson }}
+  description           = "${var.environment}-elasticache-celery-broker"
+}
+{% endif %}
+
 {% if r53_private_zone %}
 module "r53-private-zone-create" {
   source      = "./modules/r53-private-zone-create-update"

src/commcare_cloud/environment/schemas/terraform.py

@@ -37,6 +37,7 @@ class TerraformConfig(jsonobject.JsonObject):
     pgbouncer_nlbs = jsonobject.ListProperty(lambda: PgbouncerNlbs)
     internal_albs = jsonobject.ListProperty(lambda: InternalAlbs)
     elasticache_cluster = jsonobject.ObjectProperty(lambda: ElasticacheClusterConfig, default=None)
+    elasticache_celery_broker = jsonobject.ObjectProperty(lambda: ElasticacheClusterConfig, default=None)
     r53_private_zone = jsonobject.ObjectProperty(lambda: RoutePrivateZoneConfig, default=None)
     efs_file_systems = jsonobject.ListProperty(lambda: EfsFileSystem, default=None)
     ec2_auto_recovery = jsonobject.ListProperty(lambda: Ec2AutoRecovery, default=None)