diff --git a/.gitea/workflows/default.yaml b/.gitea/workflows/default.yaml index 7151b1f..efb9e33 100644 --- a/.gitea/workflows/default.yaml +++ b/.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 --exit-zero + run: flake8 ./code --benchmark - name: Type check run: mypy ./code \ No newline at end of file diff --git a/.gitea/workflows/release.yaml b/.gitea/workflows/release.yaml index 779f1b9..e515794 100644 --- a/.gitea/workflows/release.yaml +++ b/.gitea/workflows/release.yaml @@ -22,7 +22,7 @@ jobs: pip install flake8 mypy pytest if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - name: PEP8 check - run: flake8 ./code --benchmark --exit-zero + run: flake8 ./code --benchmark - name: Type check run: mypy ./code publish: diff --git a/README.md b/README.md index 0ceeb97..60977f1 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,21 @@ # 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' +``` diff --git a/code/database.py b/code/database.py index 2123f59..5eb535f 100644 --- a/code/database.py +++ b/code/database.py @@ -3,6 +3,7 @@ from dotenv import load_dotenv from configparser import ConfigParser from pathlib import Path import logging +import os # load default values @@ -21,6 +22,20 @@ 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] diff --git a/code/defaults.ini b/code/defaults.ini index 3d0ba6b..c4677b8 100644 --- a/code/defaults.ini +++ b/code/defaults.ini @@ -1,8 +1,9 @@ [main] -logger_level=info +logger_level=debug user_id=0 replicates=3 -trigger_interval_seconds=60 +trigger_interval_seconds=3600 +data_queue_flush_interval_seconds=600 [database] db_ip_address=192.168.1.2 diff --git a/code/initialise_app.py b/code/initialise_app.py index 2307209..d16a0c8 100644 --- a/code/initialise_app.py +++ b/code/initialise_app.py @@ -20,13 +20,18 @@ 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 @@ -58,6 +63,7 @@ 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__': diff --git a/code/main.py b/code/main.py index 5b84253..ee07606 100644 --- a/code/main.py +++ b/code/main.py @@ -14,45 +14,28 @@ 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 event( - replicates: int = REPLICATES, - user_id: int = USER_ID -): - logging.debug('started event') - # load environment variables - 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 +def flush_data( + data_queue: Queue, + user_id: int +) -> None: + # connect to database db = connect() + # flush queue + queue_size = data_queue.qsize() for i in range(queue_size): # get data data = data_queue.get() @@ -78,26 +61,78 @@ def event( 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() + 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 - try: - event() - except Exception as e: - logging.error(e) - else: - logging.info('finished event') + event( + data_queue=data_queue, + user_id=user_id, + replicates=replicates + ) + # extra data flush + 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)