diff --git a/.deploy/deploy-dev.yaml b/.deploy/deploy-dev.yaml index 227877a..f59cd31 100644 --- a/.deploy/deploy-dev.yaml +++ b/.deploy/deploy-dev.yaml @@ -22,6 +22,7 @@ services: networks: - configurator - queues-development + - monitoring environment: STAGE: "development" command: mailbox @@ -38,3 +39,5 @@ networks: external: true queues-development: external: true + monitoring: + external: true diff --git a/.deploy/deploy-prod.yaml b/.deploy/deploy-prod.yaml index 0043ede..535826f 100644 --- a/.deploy/deploy-prod.yaml +++ b/.deploy/deploy-prod.yaml @@ -22,6 +22,7 @@ services: networks: - configurator - queues + - monitoring environment: STAGE: "production" command: mailbox @@ -38,3 +39,5 @@ networks: external: true queues: external: true + monitoring: + external: true diff --git a/Dockerfile b/Dockerfile index 5375a3d..cd17104 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,4 +5,6 @@ COPY requirements.txt requirements.txt RUN pip install -r requirements.txt COPY . . ENV PYTHONUNBUFFERED 1 -ENTRYPOINT ["python", "main.py"] +RUN mkdir /usr/src/metrics +RUN chmod 777 run.sh +ENTRYPOINT ["./run.sh"] diff --git a/daemons/metrics.py b/daemons/metrics.py new file mode 100644 index 0000000..465de1b --- /dev/null +++ b/daemons/metrics.py @@ -0,0 +1,22 @@ +import os + +from requests import post +import json + +from daemons import base + + +class Daemon(base.Daemon): + + def execute(self): + while True: + for file in os.listdir('/usr/src/metrics'): + data = open(f'/usr/src/metrics/{file}', 'r').read() + payload = json.loads(data) + resp = post('http://monitoring:1237/api/v1/metrics/task', json=payload) + if resp.status_code == 202: + print("Metric ok") + else: + print(f'metric not ok: {resp.status_code}') + os.remove(f'/usr/src/metrics/{file}') + break diff --git a/main.py b/main.py index b1f1648..b32aa51 100644 --- a/main.py +++ b/main.py @@ -11,6 +11,9 @@ if __name__ == '__main__': elif arg == 'mailbox': print("mailbox is starting") from daemons.mailbox import Daemon + elif arg == 'metrics': + print("metrics is starting") + from daemons.metrics import Daemon else: raise ValueError(f"Unknown param {arg}") diff --git a/run.sh b/run.sh new file mode 100644 index 0000000..1826c2e --- /dev/null +++ b/run.sh @@ -0,0 +1,13 @@ +#!/bin/bash + +# Start the first process +python3 main.py $1 & + +# Start the second process +python3 main.py metrics & + +# Wait for any process to exit +wait -n + +# Exit with status of process that exited first +exit $? \ No newline at end of file diff --git a/utils/queues.py b/utils/queues.py index dc0f219..5adf3f4 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -1,4 +1,8 @@ +import datetime +import json import os +import uuid +import zoneinfo import requests import time @@ -15,6 +19,18 @@ class QueuesException(Exception): class TasksHandlerMixin: + def _send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool): + with open(f'/usr/src/{uuid.uuid4()}.json', 'w') as fp: + fp.write(json.dumps({ + 'service': 'botalka', + 'queue': self.queue_name, + 'success': success, + 'timestamp': start.strftime("%Y-%m-%dT%H:%M:%S") + "Z", + "success": success, + "execution_time_ms": (end - start).microseconds // 1000, + "environment": stage, + })) + def poll(self): while True: try: @@ -27,17 +43,24 @@ class TasksHandlerMixin: if not task: time.sleep(0.2) continue + start = datetime.datetime.now(zoneinfo.ZoneInfo("Europe/Moscow")) try: + print(f'process task with id {task["id"]}, attempt {task["attempt"]}') self.process(task['payload']) + success = True except Exception as exc: print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}') - continue - try: - resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']}) - if resp.status_code != 202: - raise QueuesException - except: - print(f'Failed to finish task id={task["id"]}') + success = False + end = datetime.datetime.now(zoneinfo.ZoneInfo("Europe/Moscow")) + if success: + try: + resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']}) + if resp.status_code != 202: + raise QueuesException + print(f'finish task with id {task["id"]}') + except: + print(f'Failed to finish task id={task["id"]}') + self._send_metric(start, end, success) @property def queue_name(self):