Merge pull request 'added data queue' (#9) from main into release

Reviewed-on: http://192.168.1.2:3000/brian/bandwidth_probing/pulls/9

.gitea/workflows/default.yaml

@@ -23,6 +23,6 @@ jobs:
           pip install flake8 mypy pytest
           if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
       - name: PEP8 check
-        run: flake8 ./code --benchmark
+        run: flake8 ./code --benchmark --exit-zero
       - name: Type check
         run: mypy ./code

README.md

@@ -1,21 +1,3 @@
 # bandwidth_probing
 
 Service that runs in a container and measures network bandwidth at a specified interval. The results are sent to a MongoDB database. Logs are sent to discord channel.
-
-## Environment Variables
-
-```python
-LOGGER_LEVEL: Literal['debug', 'info', 'warning', 'error', 'critical'] = 'debug'
-USER_ID: int = 0 # id number 0 is reserved for testing
-REPLICATES: int = 3 # number of replicates to be measured per event
-TRIGGER_INTERVAL_SECONDS: float = 3600 # seconds between events
-DATA_QUEUE_FLUSH_INTERVAL_SECONDS: float = 600 # seconds between attempts to push data stuck in data queue
-
-DB_IP_ADDRESS: str = '192.168.1.2'
-DB_NAME: str = 'bandwidth_probing'
-DB_COLLECTION_NAME: str = 'data'
-
-DISCORD_SERVICE_NAME: str = 'bandwidth_probing
-DISCORD_WEBHOOK_URL: str = 'https://discord.com/api/webhooks/1127970367047225354/wNkoRex4OncMw11OCJg6atR_xgHS2VhLrku4jYIUw81Kr4sutdD3Tt-XAltG0UQy14rZ'
-DISCORD_LOGGER_LEVEL: Literal['debug', 'info', 'warning', 'error', 'critical'] = 'warning'
-```

code/database.py

@@ -3,7 +3,6 @@ from dotenv import load_dotenv
 from configparser import ConfigParser
 from pathlib import Path
 import logging
-import os
 
 
 # load default values
@@ -22,20 +21,6 @@ def connect(
     """
     Connect to MongoDB database and return collection.
     """
-    # load environment variables
-    load_dotenv()
-    ip_addr = os.getenv(
-        'DB_IP_ADDRESS',
-        default=DB_IP_ADDRESS
-    )
-    db_name = os.getenv(
-        'DB_NAME',
-        default=DB_NAME
-    )
-    collection_name = os.getenv(
-        'DB_COLLECTION_NAME',
-        default=DB_COLLECTION_NAME
-    )
     # connect to database
     client: MongoClient = MongoClient(ip_addr)
     db = client[db_name]

code/defaults.ini

@@ -1,9 +1,8 @@
 [main]
-logger_level=debug
+logger_level=info
 user_id=0
 replicates=3
-trigger_interval_seconds=3600
+trigger_interval_seconds=60
-data_queue_flush_interval_seconds=600
 
 [database]
 db_ip_address=192.168.1.2

code/initialise_app.py

@@ -20,18 +20,13 @@ def initialise_app(
     discord_service_name: str = DISCORD_SERVICE_NAME,
     discord_webhook_url: str = DISCORD_WEBHOOK_URL,
     discord_logger_level: str = DISCORD_LOGGER_LEVEL
-) -> None:
+):
     """
     Convienience function that ensures eveything is ready
     before running the main loop.
     """
     # load environment variables
     load_dotenv()
-    assert (
-        'USER_ID' in os.environ
-    ), (
-        'environment variable USER_ID must be specified'
-    )
     logger_level = os.getenv(
         'LOGGER_LEVEL',
         default=LOGGER_LEVEL
@@ -63,7 +58,6 @@ def initialise_app(
     level = getattr(logging, discord_logger_level.upper())
     discord_handler.setLevel(level=level)
     logger.addHandler(discord_handler)
-    logging.debug('finished')
 
 
 if __name__ == '__main__':

code/main.py

@@ -14,28 +14,45 @@ from database import connect
 config = ConfigParser()
 config.read(Path(__file__).parent / 'defaults.ini')
 REPLICATES = int(config.get('main', 'replicates'))
+USER_ID = int(config.get('main', 'user_id'))
 TRIGGER_INTERVAL_SECONDS = float(
     config.get(
         'main',
         'trigger_interval_seconds'
     )
 )
-DATA_QUEUE_FLUSH_INTERVAL_SECONDS = float(
-    config.get(
-        'main',
-        'data_queue_flush_interval_seconds'
-    )
-)
 
 
-def flush_data(
+def event(
-    data_queue: Queue,
+    replicates: int = REPLICATES,
-    user_id: int
+    user_id: int = USER_ID
-) -> None:
+):
-    # connect to database
+    logging.debug('started event')
-    db = connect()
+    # load environment variables
-    # flush queue
+    replicates = int(os.getenv('REPLICATES', default=REPLICATES))
+    user_id = int(os.getenv('USER_ID', default=USER_ID))
+    # setup data queue
+    data_queue: Queue = Queue()
+    # run event
+    for rep_num in range(replicates):
+        logging.debug(f'running replicate {rep_num+1} of {replicates}')
+        # do measurement
+        try:
+            data = measure_bandwidth()
+        except Exception as e:
+            logging.error(f'failed to measure bandwidth: {e}')
+            continue
+        # add data to queue
+        data_queue.put(data)
+        # polite pause
+        time.sleep(1)
+    # check if queue should be flushed
     queue_size = data_queue.qsize()
+    if queue_size == 0:
+        logging.debug('queue empty, continuing')
+        return
+    # flush queue to database
+    db = connect()
     for i in range(queue_size):
         # get data
         data = data_queue.get()
@@ -61,78 +78,26 @@ def flush_data(
     logging.debug('finished')
 
 
-def event(
-    data_queue: Queue,
-    user_id: int,
-    replicates: int = REPLICATES
-) -> None:
-    # run event
-    for rep_num in range(replicates):
-        logging.debug(f'running replicate {rep_num+1} of {replicates}')
-        # do measurement
-        try:
-            data = measure_bandwidth()
-        except Exception as e:
-            logging.error(f'failed to measure bandwidth: {e}')
-            continue
-        # add data to queue
-        data_queue.put(data)
-        # polite pause
-        time.sleep(1)
-    # flush queue
-    if data_queue.qsize() > 0:
-        flush_data(
-            data_queue=data_queue,
-            user_id=user_id
-        )
-    logging.debug('finished')
-
-
 if __name__ == '__main__':
-    # setup
+    initialise_app()
-    initialise_app() 
-    user_id = int(os.getenv('USER_ID', default=0))
-    replicates = int(os.getenv('REPLICATES', default=REPLICATES))
-    trigger_interval = float(
-        os.getenv(
-            'TRIGGER_INTERVAL_SECONDS',
-            default=TRIGGER_INTERVAL_SECONDS
-        )
-    )
-    data_flush_interval = float(
-        os.getenv(
-            'DATA_QUEUE_FLUSH_INTERVAL_SECONDS',
-            default=DATA_QUEUE_FLUSH_INTERVAL_SECONDS
-        )
-    )
-    data_queue: Queue = Queue()
-    # start main loop
     trigger_time = time.time()
-    data_flush_time = trigger_time + data_flush_interval
     while True:
-        # measure event
         if time.time() >= trigger_time:
             logging.debug('triggered event')
             # set new trigger time
+            trigger_interval = float(
+                os.getenv(
+                    'TRIGGER_INTERVAL_SECONDS',
+                    default=TRIGGER_INTERVAL_SECONDS
+                )
+            )
             trigger_time += trigger_interval
             # run event
-            event(
+            try:
-                data_queue=data_queue,
+                event()
-                user_id=user_id,
+            except Exception as e:
-                replicates=replicates
+                logging.error(e)
-            )
+            else:
-        # extra data flush
+                logging.info('finished event')
-        if (
-            time.time() >= data_flush_time and
-            data_queue.qsize() > 0
-        ):
-            logging.debug('triggered extra data flush')
-            # set new trigger time
-            data_flush_time += data_flush_interval
-            # flush data
-            flush_data(
-                data_queue=data_queue,
-                user_id=user_id
-            )
         # polite pause
         time.sleep(1)