Merge pull request 'added separate flush interval' (#13) from main into release

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

.gitea/workflows/default.yaml

@@ -0,0 +1,28 @@
+name: pipeline
+run-name: ${{ gitea.actor }} is running script
+runs-on: ubuntu-latest
+on:
+  push:
+    branches-ignore:
+      - "release"
+image: python:3.10
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python 3.10
+        uses: https://github.com/actions/setup-python@v3
+        with:
+          python-version: "3.10"
+          architecture: "x64"
+      - name: Install Packages
+        run: |
+          python -m pip install --upgrade pip
+          pip install flake8 mypy pytest
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+      - name: PEP8 check
+        run: flake8 ./code --benchmark
+      - name: Type check
+        run: mypy ./code

.gitea/workflows/pipeline.yaml

@@ -1,59 +0,0 @@
-name: pipeline
-run-name: ${{ gitea.actor }} is running script
-runs-on: ubuntu-latest
-on: [push]
-image: python:3.10
-jobs:
-  # demo:
-  #   runs-on: ubuntu-latest
-  #   steps:
-  #     - run: echo "🎉 The job was automatically triggered by a ${{ gitea.event_name }} event."
-  #     - run: echo "🐧 This job is now running on a ${{ runner.os }} server hosted by Gitea!"
-  #     - run: echo "🔎 The name of your branch is ${{ gitea.ref }} and your repository is ${{ gitea.repository }}."
-  #     - name: Check out repository code
-  #       uses: actions/checkout@v3
-  #     - run: echo "💡 The ${{ gitea.repository }} repository has been cloned to the runner."
-  #     - run: echo "🖥️ The workflow is now ready to test your code on the runner."
-  #     - name: List files in the repository
-  #       run: |
-  #         ls ${{ gitea.workspace }}
-  #     - run: echo "🍏 This job's status is ${{ job.status }}."
-  test:
-    name: Test
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - name: Set up Python 3.10
-        uses: https://github.com/actions/setup-python@v3
-        with:
-          python-version: "3.10"
-          architecture: "x64"
-      - name: Install Packages
-        run: |
-          python -m pip install --upgrade pip
-          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
-      - name: Type check
-        run: |
-          mypy ./code
-  deploy:
-    name: Deploy
-    runs-on: ubuntu-latest
-    needs: [test]
-    if: gitea.ref == 'refs/heads/release'
-    env:
-      repository_host: 192.168.1.2:5000
-      repository_name: ${{ gitea.repository }}
-      repository_tag: latest
-    steps:
-      - uses: actions/checkout@v3
-      - name: Build
-        run: docker build . --file Dockerfile --tag ${repository_name}:${repository_tag}
-      - name: Tag
-        run: docker tag ${repository_name}:${repository_tag} ${repository_host}/${repository_name}:${repository_tag}
-      - name: Push
-        run: docker push ${repository_host}/${repository_name}:${repository_tag}
-

.gitea/workflows/release.yaml

@@ -0,0 +1,52 @@
+name: Docker
+run-name: Build and Publish Service
+runs-on: ubuntu-latest
+on:
+  push:
+    tags:
+      - 'v[0-9]+.[0-9]+.[0-9]+'
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python 3.10
+        uses: https://github.com/actions/setup-python@v3
+        with:
+          python-version: "3.10"
+          architecture: "x64"
+      - name: Install Packages
+        run: |
+          python -m pip install --upgrade pip
+          pip install flake8 mypy pytest
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+      - name: PEP8 check
+        run: flake8 ./code --benchmark
+      - name: Type check
+        run: mypy ./code
+  publish:
+    name: Publish
+    runs-on: ubuntu-latest
+    needs: [test]
+    steps:
+      - uses: actions/checkout@v3
+      -
+        name: Set Environment Variables
+        id: vars
+        env:
+          repository: ${{gitea.repository}}
+          ref_name: ${{gitea.ref_name}}
+        run: |
+          echo $repository
+          echo ::set-output name=OWNER::${repository#/*}
+          echo ::set-output name=PROJECT::${repository#*/}
+          echo $ref_name
+          echo ::set-output name=TAG::${ref_name#refs/*/}
+      -
+        name: Push Image
+        uses: docker/build-push-action@v2
+        with:
+          context: .
+          push: true
+          tags: 192.168.1.2:5000/${{gitea.repository}}:${{steps.vars.outputs.TAG}}

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'
+```

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]

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

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__':

code/main.py

@@ -1,8 +1,10 @@
 import time
 import logging
 import os
+import json
 from configparser import ConfigParser
 from pathlib import Path
+from queue import Queue
 from initialise_app import initialise_app
 from bandwidth import measure as measure_bandwidth
 from database import connect
@@ -12,67 +14,125 @@ 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(
+def flush_data(
-    replicates: int = REPLICATES,
+    data_queue: Queue,
-    user_id: int = USER_ID
+    user_id: int
-):
+) -> None:
-    logging.debug('started event')
+    # connect to database
-    # load environment variables
-    replicates = int(os.getenv('REPLICATES', default=REPLICATES))
-    user_id = int(os.getenv('USER_ID', default=USER_ID))
-    # setup database connection
     db = connect()
-    # run event
+    # flush queue
-    for rep_num in range(replicates):
+    queue_size = data_queue.qsize()
-        logging.info(f'running replicate {rep_num}')
+    for i in range(queue_size):
-        # do measurement
+        # get data
-        try:
+        data = data_queue.get()
-            res = measure_bandwidth()
+        # prepare payload
-        except Exception as e:
-            logging.error(f'failed to measure bandwidth: {e}')
-            continue
-        # upload result to database
         payload_dict = {
             'user_id': user_id,
-            'data': res
+            'data': data
         }
+        # send to database
         try:
             db_id = db.insert_one(payload_dict).inserted_id
         except Exception as e:
-            logging.error(f'failed sending results to database: {e}')
+            logging.error('failed sending data to database')
+            logging.info(
+                'failed pushing to database:\n'
+                f'{json.dumps(payload_dict, indent=4)}\n'
+                f'with error: {e}'
+            )
+            data_queue.put(data)    # put data back in queue
             continue
-        logging.debug(f'data sent to database received db_id: {db_id}')
+        else:
+            logging.debug(f'data sent to database received id: {db_id}')
+    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()
-    trigger_time = time.time()
+    user_id = int(os.getenv('USER_ID', default=0))
-    while True:
+    replicates = int(os.getenv('REPLICATES', default=REPLICATES))
-        if time.time() >= trigger_time:
-            # set new trigger time
     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_time += trigger_interval
             # run event
-            try:
+            event(
-                event()
+                data_queue=data_queue,
-            except Exception as e:
+                user_id=user_id,
-                logging.error(e)
+                replicates=replicates
-            else:
+            )
-                logging.info('finished event')
+        # 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)