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

@@ -26,27 +26,3 @@ jobs:
         run: flake8 ./code --benchmark --exit-zero
       - name: Type check
         run: mypy ./code
-  # publish:
-  #   name: Publish
-  #   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_version: v1.0
-  #   steps:
-  #     - uses: actions/checkout@v3
-  #     - name: Build
-  #       run: |
-  #         echo "> docker build . --file Dockerfile --tag ${repository_name}:${repository_version}"
-  #         docker build . --file Dockerfile --tag ${repository_name}:${repository_version}
-  #     - name: Tag
-  #       run: |
-  #         echo "> docker tag ${repository_name}:${repository_version} ${repository_host}/${repository_name}:${repository_version}"
-  #         docker tag ${repository_name}:${repository_version} ${repository_host}/${repository_name}:${repository_version}
-  #     - name: Push
-  #       run: |
-  #         echo "> docker push ${repository_host}/${repository_name}:${repository_version}"
-  #         docker push ${repository_host}/${repository_name}:${repository_version}
-

.gitea/workflows/release.yaml

@@ -39,18 +39,14 @@ jobs:
           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: Show Environment Variables
-        env:
-          tag: ${{steps.vars.output.TAG}}
-        run: |
-          echo $tag
       -
         name: Push Image
         uses: docker/build-push-action@v2
         with:
           context: .
           push: true
-          tags: 192.168.1.2:5000/${{gitea.repository}}:${{steps.get_version.outputs.VERSION}}
+          tags: 192.168.1.2:5000/${{gitea.repository}}:${{steps.vars.outputs.TAG}}

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
@@ -29,29 +31,50 @@ def event(
     # 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()
+    # setup data queue
+    data_queue: Queue = Queue()
     # run event
     for rep_num in range(replicates):
-        logging.info(f'running replicate {rep_num}')
+        logging.debug(f'running replicate {rep_num+1} of {replicates}')
         # do measurement
         try:
-            res = measure_bandwidth()
+            data = measure_bandwidth()
         except Exception as e:
             logging.error(f'failed to measure bandwidth: {e}')
             continue
-        # upload result to database
+        # 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()
+        # prepare payload
         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}')
-        time.sleep(1)
+        else:
+            logging.debug(f'data sent to database received id: {db_id}')
     logging.debug('finished')
 
 
@@ -60,6 +83,7 @@ if __name__ == '__main__':
     trigger_time = time.time()
     while True:
         if time.time() >= trigger_time:
+            logging.debug('triggered event')
             # set new trigger time
             trigger_interval = float(
                 os.getenv(
@@ -75,4 +99,5 @@ if __name__ == '__main__':
                 logging.error(e)
             else:
                 logging.info('finished event')
+        # polite pause
         time.sleep(1)