Further configuration

Author: YuryHrytsuk

services/rabbit/README.md

@@ -7,6 +7,12 @@ rabbitmq.conf and advanced.config changes take effect after a node restart. This
 
 Source: https://www.rabbitmq.com/docs/next/configure#config-changes-effects
 
+## How to add / remove nodes
+
+The only supported way, is to completely shutdown the cluster (docker stack and most likely rabbit node volumes) and start brand new. 
+
+With manual effort this can be done on the running cluster, by adding 1 more rabbit node manually (as a separate docker stack or new service) and manually executing rabbitmqctl commands (some hints can be found here https://www.rabbitmq.com/docs/clustering#creating)
+
 ## Enable node Maintenance mode
 
 1. Get inside container's shell (`docker exec -it <container-id> bash`)

services/rabbit/docker-compose.yml.j2

@@ -3,9 +3,14 @@
 services:
   loadbalancer:
     image: haproxy:3.2
-    ports:
-      - "5672:5672"
-      - "15672:15672"
+    deploy:
+      # https://discourse.haproxy.org/t/haproxy-high-availability-configuration/11983
+      replicas: 2
+      # necessary to preserve client ip
+      # otherwise we see overlay network lb ip
+      # (rabbitmq management dashboard connection section)
+      endpoint_mode: dnsrr
+    # healthcheck: https://github.com/haproxy/haproxy/issues/3091
     networks:
       - rabbit
     configs:

services/rabbit/haproxy.cfg.j2

@@ -3,10 +3,27 @@ global
     maxconn 4096
     pidfile /tmp/haproxy.pid
 
+# haproxy by default resolves server hostname only once
+# this breaks if container restarts. By using resolvers
+# we tell haproxy to re-resolve the hostname (so container
+# restarts are handled properly)
+resolvers dockerdns
+    nameserver dns1 127.0.0.11:53
+    resolve_retries 3
+    timeout resolve 1s
+    timeout retry   1s
+    hold other      10s
+    hold refused    10s
+    hold nx         10s
+    hold timeout    10s
+    hold valid      10s
+    hold obsolete   10s
+
 defaults
     log global
     mode tcp
     option tcplog
+
     timeout connect 5s
     timeout client 30s
     timeout server 30s
@@ -21,16 +38,20 @@ frontend rabbit_dashboard
 
 backend rabbit_backends
     balance roundrobin
-    server rabbit01 rabbit_rabbit01:5672 check inter 5s rise 2 fall 3
-    server rabbit02 rabbit_rabbit02:5672 check inter 5s rise 2 fall 3
-    server rabbit03 rabbit_rabbit03:5672 check inter 5s rise 2 fall 3
+
+    # init-addrs libc,none - start even if there aren’t any running
+    server rabbit01 rabbit_rabbit01:5672 check resolvers dockerdns init-addr libc,none inter 5s rise 2 fall 3 send-proxy
+    server rabbit02 rabbit_rabbit02:5672 check resolvers dockerdns init-addr libc,none inter 5s rise 2 fall 3 send-proxy
+    server rabbit03 rabbit_rabbit03:5672 check resolvers dockerdns init-addr libc,none inter 5s rise 2 fall 3 send-proxy
 
 backend rabbit_dashboard_backends
     mode http
     balance roundrobin
+
     option forwardfor
     http-request set-header X-Forwarded-Port %[dst_port]
     http-request add-header X-Forwarded-Proto https if { ssl_fc }
-    server rabbit01 rabbit_rabbit01:15672 check inter 5s rise 2 fall 3
-    server rabbit02 rabbit_rabbit02:15672 check inter 5s rise 2 fall 3
-    server rabbit03 rabbit_rabbit03:15672 check inter 5s rise 2 fall 3
+
+    server rabbit01 rabbit_rabbit01:15672 check resolvers dockerdns init-addr libc,none inter 5s rise 2 fall 3
+    server rabbit02 rabbit_rabbit02:15672 check resolvers dockerdns init-addr libc,none inter 5s rise 2 fall 3
+    server rabbit03 rabbit_rabbit03:15672 check resolvers dockerdns init-addr libc,none inter 5s rise 2 fall 3

services/rabbit/rabbitmq.conf.j2

@@ -12,3 +12,7 @@ cluster_formation.classic_config.nodes.{{ ix }} = rabbit@rabbit_rabbit0{{ ix }}
 ## at declaration time.
 # https://www.rabbitmq.com/docs/quorum-queues#quorum-requirements
 quorum_queue.initial_cluster_size = {{ RABBIT_QUORUM_QUEUE_DEFAULT_REPLICA_COUNT }}
+
+# Extract proper client ip when behind a proxy (e.g. haproxy)
+# https://www.rabbitmq.com/docs/networking#proxy-protocol
+proxy_protocol = true

services/rabbit/test/pub.py

@@ -10,21 +10,12 @@
 )
 logger = logging.getLogger(__name__)
 
-hosts = os.environ["RABBIT_HOSTS"].split(",")
-credentials = pika.PlainCredentials(os.getenv("RABBIT_USER"), os.getenv("RABBIT_PASS"))
-
-endpoints = [
-    pika.URLParameters(
-        f"amqp://{credentials.username}:{credentials.password}@{h.strip()}:5672/"
-    )
-    for h in hosts
-]
+host = os.environ["RABBIT_HOSTS"].strip()
+credentials = pika.PlainCredentials(os.environ["RABBIT_USER"], os.environ["RABBIT_PASS"])
+parameters = pika.ConnectionParameters(host=host, port=5672, credentials=credentials, client_properties={"connection_name": "publisher"})
 
 while True:
-    logger.info(f"Verbinde zu RabbitMQ Hosts: {hosts}")
-
-    random.shuffle(endpoints)
-    connection = pika.BlockingConnection(endpoints)
+    connection = pika.BlockingConnection(parameters)
     try:
         channel = connection.channel()
 
@@ -43,7 +34,8 @@
             )
             logger.info(f"Gesendet: {msg}")
             time.sleep(3)
-    except pika.exceptions.ConnectionClosedByBroker:
+    except Exception:
         logger.error(
             "Verbindung zum RabbitMQ Broker wurde geschlossen. Versuche erneut zu verbinden..."
         )
+        time.sleep(5)

services/rabbit/test/sub.py

@@ -1,33 +1,23 @@
 import logging
 import os
 import random
-
+import time
 import pika
 
 logging.basicConfig(
     level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s"
 )
 logger = logging.getLogger(__name__)
 
-hosts = os.environ["RABBIT_HOSTS"].split(",")
-credentials = pika.PlainCredentials(os.getenv("RABBIT_USER"), os.getenv("RABBIT_PASS"))
-
-hosts = os.environ["RABBIT_HOSTS"].split(",")
-credentials = pika.PlainCredentials(os.getenv("RABBIT_USER"), os.getenv("RABBIT_PASS"))
-
-endpoints = [
-    pika.URLParameters(
-        f"amqp://{credentials.username}:{credentials.password}@{h.strip()}:5672/"
-    )
-    for h in hosts
-]
+host = os.environ["RABBIT_HOSTS"].strip()
+credentials = pika.PlainCredentials(os.environ["RABBIT_USER"], os.environ["RABBIT_PASS"])
+parameters = pika.ConnectionParameters(host=host, port=5672, credentials=credentials, client_properties={"connection_name": "subscriber"})
 
 while True:
-    logger.info(f"Verbinde zu RabbitMQ Hosts: {hosts}")
+    logger.info(f"Verbinde zu RabbitMQ Host: {host}")
 
-    random.shuffle(endpoints)
     try:
-        connection = pika.BlockingConnection(endpoints)
+        connection = pika.BlockingConnection(parameters)
         channel = connection.channel()
 
         logger.info("Queue 'hello' (quorum) deklarieren...")
@@ -42,7 +32,8 @@ def callback(ch, method, properties, body):
         channel.basic_consume(queue="hello", on_message_callback=callback)
         logger.info("Warte auf Nachrichten...")
         channel.start_consuming()
-    except pika.exceptions.ConnectionClosedByBroker:
+    except Exception:
         logger.error(
             "Verbindung zum RabbitMQ Broker wurde geschlossen. Versuche erneut zu verbinden..."
         )
+        time.sleep(5)