Merge branch '11-zeitlich-gruppierte-update-queue' into 'main'

Resolve "Zeitlich gruppierte Update Queue"

Closes #11

See merge request imprj/01-bgp-testbed/zettabgp!9

Author: DlieBG

README.md

@@ -62,12 +62,24 @@ The process can be started from within ExaBGP.\
 The Messages will be received using the stdin pipe.
 ```
 Options:
-  -r, --no-rabbitmq
+  -d, --no-rabbitmq-direct
+  -g, --rabbitmq-grouped INTEGER  Queue group interval in minutes. [default: (5)]
   -l, --no-mongodb-log
   -s, --no-mongodb-state
   -t, --no-mongodb-statistics
 ```
 
+##### Queue Group Interval
+Besides the direct RabbitMQ queue (Exchange: `zettabgp` with Routing Key: `direct`), a grouped queue can be activated.\
+The grouped queue (Exchange: `zettabgp` with Routing Key: `grouped`) will be filled white grouped route updates when enabled with the `-g` option.\
+The group interval defaults to 5 minutes.\
+Alternative intervals can be set as an argument to option `-g`.\
+A 10 minute interval can be set as following.
+```
+zettabgp exabgp -g 10
+```
+When no `-g` option is present, no grouped updates will appear at all.
+
 #### `zettabgp mrt-simulation`
 The `mrt-simulation` subcommand is used for processing MRT files.\
 It is mendatory to provide a valid path to a mrt file.\
@@ -76,14 +88,18 @@ Arguments:
   MRT_FILE
 
 Options:
-  -r, --no-rabbitmq
+  -d, --no-rabbitmq-direct
+  -g, --rabbitmq-grouped INTEGER  Queue group interval in minutes. [default: (5)]
   -l, --no-mongodb-log
   -s, --no-mongodb-state
   -t, --no-mongodb-statistics
   -p, --playback-speed INTEGER    Playback speed in multiples of real time. [default: (1)]
-  -o, --playback-interval INTEGER Playback interval in minutes.  [default: (5)]
+  -o, --playback-interval INTEGER Playback interval in minutes. [default: (5)]  
 ```
 
+##### Queue Group Interval
+See [`exabgp` command reference](#queue-group-interval).
+
 ##### Playback Speed
 Without specifying a playback speed, `mrt-simulation` will replay all route updates at once.\
 When defining playback speed, the replay of the updates will be done in multiples of real time.\

src/adapters/rabbitmq.py

@@ -1,10 +1,11 @@
 from src.parsers.route_update import RouteUpdateParser
 from src.models.route_update import RouteUpdate
-import pika, os
+from datetime import timedelta, datetime
+import pika, json, os
 
 class RabbitMQAdapter:
     '''This class is responsible for receiving the parsed messages and forwarding them to the RabbitMQ message broker'''
-    def __init__(self, parser: RouteUpdateParser):
+    def __init__(self, parser: RouteUpdateParser, no_direct: bool, queue_interval: int):
         connection = pika.BlockingConnection(
             pika.ConnectionParameters(
                 host=os.getenv('RABBIT_MQ_HOST', 'localhost'),
@@ -15,27 +16,65 @@ def __init__(self, parser: RouteUpdateParser):
         channel = connection.channel()
         channel.exchange_declare(
             exchange='zettabgp',
-            exchange_type='fanout',
+            exchange_type='direct',
         )
 
+
         '''Declares the test_bgp_updates queue and binds it to the zettabgp exchange'''
-        def _declare_test_queue(queue_name: str):
+        def _declare_test_queue(queue_name: str, routing_key: str):
             channel.queue_declare(
                 queue=queue_name,
             )
             channel.queue_bind(
                 exchange='zettabgp',
                 queue=queue_name,
+                routing_key=routing_key,
             )
 
         _declare_test_queue(
-            queue_name='test_bgp_updates',
+            queue_name='test_direct_route_updates',
+            routing_key='direct',
+        )
+        _declare_test_queue(
+            queue_name='test_grouped_route_updates',
+            routing_key='grouped',
         )
 
-        @parser.on_update
-        def on_update(message: RouteUpdate):
-            channel.basic_publish(
-                exchange='zettabgp',
-                body=message.model_dump_json(),
-                routing_key='',
-            )
+        if not no_direct:
+            @parser.on_update
+            def direct(message: RouteUpdate):
+                channel.basic_publish(
+                    exchange='zettabgp',
+                    body=message.model_dump_json(),
+                    routing_key='direct',
+                )
+
+        if queue_interval:
+            queue_interval_stop: datetime = None
+            queue_interval_messages: list[RouteUpdate] = []
+        
+            @parser.on_update
+            def grouped(message: RouteUpdate):
+                nonlocal queue_interval_stop
+
+                if queue_interval_stop:
+                    if message.timestamp >= queue_interval_stop:
+                        channel.basic_publish(
+                            exchange='zettabgp',
+                            body=json.dumps([
+                                message.model_dump(
+                                    mode='json',
+                                ) 
+                                    for message in queue_interval_messages
+                            ]),
+                            routing_key='grouped',
+                        )
+
+                        queue_interval_stop = queue_interval_stop + timedelta(minutes=queue_interval)
+                        queue_interval_messages.clear()
+                else:
+                    # -1 seconds to avoid unpublished last interval when the next message is exactly at the stop time
+                    # this occurs when simulating with mrt-simulation and same -o value
+                    queue_interval_stop = message.timestamp + timedelta(minutes=queue_interval, seconds=-1)
+
+                queue_interval_messages.append(message)

src/main.py

@@ -16,10 +16,20 @@ def cli():
     help='Process ExaBGP messages.',
 )
 @click.option(
-    '--no-rabbitmq',
-    '-r',
+    '--no-rabbitmq-direct',
+    '-d',
     is_flag=True,
 )
+@click.option(
+    '--rabbitmq-grouped',
+    '-g',
+    type=int,
+    default=None,
+    show_default='5',
+    is_flag=False,
+    flag_value=5,
+    help='Queue group interval in minutes.',
+)
 @click.option(
     '--no-mongodb-log',
     '-l',
@@ -35,12 +45,14 @@ def cli():
     '-t',
     is_flag=True,
 )
-def exabgp(no_rabbitmq: bool, no_mongodb_log: bool, no_mongodb_state: bool, no_mongodb_statistics: bool):
+def exabgp(no_rabbitmq_direct: bool, rabbitmq_grouped: int, no_mongodb_log: bool, no_mongodb_state: bool, no_mongodb_statistics: bool):
     parser = ExaBGPParser()
 
-    if not no_rabbitmq:
+    if not no_rabbitmq_direct or rabbitmq_grouped:
         RabbitMQAdapter(
             parser=parser,
+            no_direct=no_rabbitmq_direct,
+            queue_interval=rabbitmq_grouped,
         )
 
     if not no_mongodb_log or not no_mongodb_state or not no_mongodb_statistics:
@@ -64,10 +76,20 @@ def exabgp(no_rabbitmq: bool, no_mongodb_log: bool, no_mongodb_state: bool, no_m
     help='Process BGP4MP MRT files.',
 )
 @click.option(
-    '--no-rabbitmq',
-    '-r',
+    '--no-rabbitmq-direct',
+    '-d',
     is_flag=True,
 )
+@click.option(
+    '--rabbitmq-grouped',
+    '-g',
+    type=int,
+    default=None,
+    show_default='5',
+    is_flag=False,
+    flag_value=5,
+    help='Queue group interval in minutes.',
+)
 @click.option(
     '--no-mongodb-log',
     '-l',
@@ -110,12 +132,14 @@ def exabgp(no_rabbitmq: bool, no_mongodb_log: bool, no_mongodb_state: bool, no_m
         resolve_path=True,
     ),
 )
-def mrt_simulation(no_rabbitmq: bool, no_mongodb_log: bool, no_mongodb_state: bool, no_mongodb_statistics: bool, playback_speed: int, playback_interval: int, mrt_file: str):
+def mrt_simulation(no_rabbitmq_direct: bool, rabbitmq_grouped: int, no_mongodb_log: bool, no_mongodb_state: bool, no_mongodb_statistics: bool, playback_speed: int, playback_interval: int, mrt_file: str):
     parser = MrtBgp4MpParser()
 
-    if not no_rabbitmq:
+    if not no_rabbitmq_direct or rabbitmq_grouped:
         RabbitMQAdapter(
             parser=parser,
+            no_direct=no_rabbitmq_direct,
+            queue_interval=rabbitmq_grouped,
         )
 
     if not no_mongodb_log or not no_mongodb_state or not no_mongodb_statistics: