feat: requeue with dynamic delay

Author: linuxbandit

dispatcher/Makefile

@@ -11,7 +11,7 @@ down:
 	COMPOSE_PROJECT_NAME=myaegee docker-compose -f ./docker/docker-compose.yml -f ./docker/docker-compose.dev.yml down
 
 test: build-dev
-	@sleep 5; sed -i "s/'172.18.0.2'/$$(docker inspect myaegee_rabbit_1 | jq .[0].NetworkSettings.Networks.OMS.IPAddress)/" helpers/send.py ; cd helpers && python send.py
+	@sleep 5; sed -i "s/'172.18.0.X'/$$(docker inspect myaegee_rabbit_1 | jq .[0].NetworkSettings.Networks.OMS.IPAddress)/" helpers/send.py ; cd helpers && python send.py
 
 rabbit:
 	@DOCKER_BUILDKIT=0 docker build -t aegee/rabbit:latest -f docker/Dockerfile.rabbit . && docker push aegee/rabbit

dispatcher/README.md

@@ -44,3 +44,15 @@ rather in order:
 1. [x] When RabbitMQ quits or crashes it will forget the queues and messages unless you tell it not to: we need to mark both the queue and messages as durable
 1. [ ] Add auto-retry (DLQ). rabbit is smart and doesn't let me process a message again unless i force it.. https://devcorner.digitalpress.blog/rabbitmq-retries-the-new-full-story/
 1. [ ] add the telegram queue
+1. investigate the massmailer queue: a queue which picks every message, and creates a list of "bcc" to send only one email? (danger: queue needs something like batch ack..) - OR it is not feasible at all because "mass"mailer is still "personalised" mailer?
+
+1. why do we even have a `<`title`>` (which is dynamic), why not using directly the subject? (re: the body of the email)
+1. remove extension Jinja2 (into jinja)
+1. make it such that templates list is read from fs (for dynamic tests)
+
+
+
+https://www.rabbitmq.com/publishers.html#unroutable
+
+
+Each consumer (subscription) has an identifier called a consumer tag. It can be used to unsubscribe from messages. Consumer tags are just strings.

dispatcher/dispatcher/main.py

@@ -54,12 +54,48 @@
                    queue='requeue_queue',
                    routing_key='wait')
 
+def requeue_wait(ch, method, properties, body, reason):
+    REQUEUE_DELAY_DURATIONS = [
+        5 * 60000, # 5 mins
+        50 * 60000, # 50 mins
+        5*60 * 60000, # 5 hrs
+        5*60*10 * 60000, # 50 hrs
+        5*60*20 * 60000, # 100 hrs
+    ]
+
+    if (properties.headers != None and "x-delay" in properties.headers):
+        index = REQUEUE_DELAY_DURATIONS.index(int(properties.headers["x-delay"]))
+        if (index+1 == len(REQUEUE_DELAY_DURATIONS) ):
+            print('Max retry time hit, dropping message')
+            # TODO: notify someone that they've been sloppy
+            ch.basic_ack(delivery_tag = method.delivery_tag)
+            return
+        else:
+            print(f'Attempt {index+1}/{len(REQUEUE_DELAY_DURATIONS)-1}')
+            wait = REQUEUE_DELAY_DURATIONS[index+1]
+    else:
+        wait = REQUEUE_DELAY_DURATIONS[0]
+
+    headers = {
+        'reason': reason,
+        'x-delay': wait,
+    }
+    prop = pika.BasicProperties(
+        headers=headers,
+        delivery_mode = pika.spec.PERSISTENT_DELIVERY_MODE,
+        )
+    channel.basic_publish(exchange='wait_exchange',
+                        routing_key='wait',
+                        body=body,
+                        properties=prop) #NOTE it completely ignores the previous properties (and it's fine)
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
 def send_email(ch, method, properties, body):
     """
     Callback for the NORMAL MESSAGE
     Output: send an email
         OR
-    Output: Dead-letter queue
+    Output: Wait-exchange
     """
     msg = json.loads(body)
 
@@ -69,37 +105,20 @@ def send_email(ch, method, properties, body):
         # TODO: send a notification to someone about adding a template
         # NOTE: this is a requeuable message
         print(f"Template {msg['template']}.jinja2 not found")
-        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
+        requeue_wait(ch, method, properties, body, reason="template_not_found")
         return
 
     try:
         rendered = template.render(msg['parameters'], altro=msg['subject'])
     except exceptions.UndefinedError as e:
         # NOTE: this is a NON-requeuable message
         print(f"Error in rendering: some parameter is undefined (error: {e}; message: {msg})")
-# TODO: check if there is no x-header about DLQ if
-# i send directly to queue from this (i.e. without passing from DLQ),
-
-        # headers = {
-        #     'reason': 'parameter_undefined',
-        #     'x-delay': '60000',
-        # }
-        # prop = pika.BasicProperties(
-        #     headers=headers,
-        #     delivery_mode = pika.spec.PERSISTENT_DELIVERY_MODE,
-        #     )
-        # channel.basic_publish(exchange='wait_exchange',
-        #                   routing_key='wait',
-        #                   body=body,
-        #                   properties=prop)
-        # ch.basic_ack(delivery_tag=method.delivery_tag)
-
-        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
+        requeue_wait(ch, method, properties, body, reason="parameter_undefined")
         return
     except exceptions.TemplateNotFound:
         # NOTE: this is a requeuable message
         print(f"A sub-template in {msg['template']}.jinja2 was not found")
-        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
+        requeue_wait(ch, method, properties, body, reason="sub-template_not_found")
         return
 
     email = EmailMessage()
@@ -114,10 +133,12 @@ def send_email(ch, method, properties, body):
 def process_dead_letter_messages(ch, method, properties, body):
     """
     Callback for the ERROR MESSAGE
-    Output: Requeue on delayed exchange (if error about missing template)
-        OR
-    Output: Remove (if error about unretrievable data)
-    #TODO I can't know which is the case, with the current design
+    Output: none yet. I don't expect for messages to fall here, I keep the DLQ for safety
+
+    @see https://stackoverflow.com/a/58500336
+    "The way to do this is not to use NACK at all but to generate and return a 'new' message
+    (which is simply the current message you are handling, but adding new headers to it).
+    It appears that a NACK is basically doing this anyway according to the AMQP spec."
     """
     REQUEUE_DELAY_DURATIONS = [
         5 * 60000, # 5 mins
@@ -126,15 +147,9 @@ def process_dead_letter_messages(ch, method, properties, body):
         5*60*10 * 60000, # 50 hrs
         5*60*20 * 60000, # 100 hrs
     ]
-    wait_for = REQUEUE_DELAY_DURATIONS[4] # TODO make it dynamic
-
-    print(f'DLQ')
-    print(properties.headers)
-    print(f'Message was in "{properties.headers["x-first-death-exchange"]}" (specifically "{properties.headers["x-first-death-queue"]}" queue) and was rejected because: {properties.headers["x-first-death-reason"]}')
-    print()
+    wait_for = REQUEUE_DELAY_DURATIONS[-1]
 
     headers = {
-            #'reason': 'parameter_undefined',
             'x-delay': wait_for,
         }
     fullheaders = {**properties.headers, **headers}
@@ -154,18 +169,20 @@ def process_requeue(ch, method, properties, body):
     Callback for the WAITING MESSAGES
     Output: Requeue on normal exchange (if error about missing template)
         OR
-    Output: Remove (if limit time reached) #TODO
+    Output: Remove (if unfixable error)
     """
 
-    print(f'REQUEUE-WAIT')
-    print(properties.headers)
-    print(f'Message was in "{properties.headers["x-first-death-exchange"]}" (specifically "{properties.headers["x-first-death-queue"]}" queue) and was rejected because: {properties.headers["x-first-death-reason"]}')
-    print()
+    if (properties.headers["reason"] == 'parameter_undefined'):
+        print('Impossible to fix error, dropping message')
+        #TODO output something/notify to leave a trail for better debugging on what was missing
+        ch.basic_ack(delivery_tag = method.delivery_tag)
+        return
 
     channel.basic_publish(exchange='eml',
                           routing_key='mail',
                           body=body,
                           properties=pika.BasicProperties(
+                              headers = properties.headers, # propagation to avoid endless loop
                               delivery_mode = pika.spec.PERSISTENT_DELIVERY_MODE,
                           ))
     ch.basic_ack(delivery_tag = method.delivery_tag)

dispatcher/docker/docker-compose.dev.yml

@@ -2,6 +2,12 @@ version: "3.4"
 
 services:
 
+  rabbit:
+    build:
+      context: ./${PATH_DISPATCHER}/..
+      dockerfile: ./docker/Dockerfile.rabbit
+    image: aegee/rabbit:3.11-mgmt-delayxch
+
   dispatcher:
     build:
       context: ./${PATH_DISPATCHER}/..

dispatcher/docker/docker-compose.yml

@@ -3,7 +3,7 @@ version: "3.4"
 services:
 
   rabbit:
-    image: aegee/rabbit
+    image: aegee/rabbit:3.11-mgmt-delayxch
     restart: always
     expose:
       - 5672

dispatcher/helpers/send.py

@@ -12,6 +12,7 @@
 from faker import Faker
 faker = Faker()
 
+ERROR_TEST=True
 RANDOM_AMOUNT_TEST=False
 MIN_MSG=1
 MAX_MSG=8
@@ -122,7 +123,7 @@ def agora_sentence():
     },
 }
 
-RABBIT_HOST="172.18.0.2" #FIXME
+RABBIT_HOST="172.18.0.X" #FIXME (as this is a python script launched on host we cant use docker's dns)
 connection = pika.BlockingConnection(pika.ConnectionParameters(RABBIT_HOST))
 channel = connection.channel()
 
@@ -132,15 +133,15 @@ def agora_sentence():
 channel.queue_declare(queue='email',
                       arguments={
                         'x-dead-letter-exchange': "dead_letter_exchange",
-                        'x-dead-letter-routing-key': "dead_letterl_routing_key",
+                        'x-dead-letter-routing-key': "dead_letter_routing_key",
                         'x-death-header': True,
                       },
                       durable=True)
 channel.queue_bind(exchange='eml',
                    queue='email',
                    routing_key='mail')
 
-def generate_fake_payload(subj="", template=""):
+def generate_fake_payload(subj="", template="", return_malformed_mail=False):
     email = {
         "from": "[email protected]",
         "to": [faker.email() for _ in range(random.randrange(1,3))],
@@ -214,6 +215,9 @@ def generate_fake_payload(subj="", template=""):
             """
         }
     }
+    if(return_malformed_mail):
+        email["template"] = MAIL_TEMPLATES["EVENTS"]["MAIL_EVENT_UPDATED"]
+        del email["parameters"]["event"]
     return email
 
 
@@ -228,7 +232,27 @@ def generate_fake_payload(subj="", template=""):
                                 delivery_mode = pika.spec.PERSISTENT_DELIVERY_MODE
                             ))
         print(f" [x] Sent {email['subject']} (to {email['to']})")
-    print(f" Gee, I sent all  {amount} ")
+    print(f" Gee, I sent all {amount} ")
+
+if(ERROR_TEST):
+    #ERROR 1: no template
+    email = generate_fake_payload(template="notexisting", subj="will never be delivered")
+    channel.basic_publish(exchange='eml',
+                        routing_key='mail',
+                        body=json.dumps(email),
+                        properties=pika.BasicProperties(
+                            delivery_mode = pika.spec.PERSISTENT_DELIVERY_MODE
+                        ))
+    print(f" [x] Sent {email['subject']} (to {email['to']})")
+    #ERROR 2: missing field
+    email = generate_fake_payload(return_malformed_mail=True, subj="will never be delivered")
+    channel.basic_publish(exchange='eml',
+                        routing_key='mail',
+                        body=json.dumps(email),
+                        properties=pika.BasicProperties(
+                            delivery_mode = pika.spec.PERSISTENT_DELIVERY_MODE
+                        ))
+    print(f" [x] Sent {email['subject']} (to {email['to']})")
 
 if(ALL_TEMPLATES_TEST):
     templates_tested=0