From 4af857e2f3890c8a37ba93822bf436dae309acc8 Mon Sep 17 00:00:00 2001 From: Egor Matveev Date: Sun, 15 Jun 2025 05:07:16 +0300 Subject: [PATCH 1/8] fix --- .deploy/deploy-dev.yaml | 3 +++ .deploy/deploy-prod.yaml | 3 +++ utils/queues.py | 46 ++++++++++++++++++++++++++++++++++------ 3 files changed, 45 insertions(+), 7 deletions(-) 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/utils/queues.py b/utils/queues.py index dc0f219..3ec1e8b 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -1,4 +1,7 @@ +from concurrent.futures import ThreadPoolExecutor +import datetime import os +import zoneinfo import requests import time @@ -10,11 +13,35 @@ else: QUEUES_URL = 'http://queues:1239' +executor = ThreadPoolExecutor(max_workers=1) + + class QueuesException(Exception): ... class TasksHandlerMixin: + def _send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool): + try: + resp = requests.post('http://monitoring:1237/api/v1/metrics/task', json={ + '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, + }) + if resp.status_code == 202: + print("Metric ok") + else: + print(f'metric not ok: {resp.status_code}') + except Exception as e: + print(f"Error sending metric: {e}") + + def send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool): + executor.submit(self._send_metric, start, end, success) + def poll(self): while True: try: @@ -27,17 +54,22 @@ class TasksHandlerMixin: if not task: time.sleep(0.2) continue + start = datetime.datetime.now(zoneinfo.ZoneInfo("Europe/Moscow")) try: 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 + except: + print(f'Failed to finish task id={task["id"]}') + self.send_metric(start, end, success) @property def queue_name(self): From 2bff8983b54c129ebde158cb3405ed6ae0769fbf Mon Sep 17 00:00:00 2001 From: Egor Matveev Date: Sun, 15 Jun 2025 05:11:23 +0300 Subject: [PATCH 2/8] fix --- utils/queues.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/utils/queues.py b/utils/queues.py index 3ec1e8b..a571358 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -40,7 +40,8 @@ class TasksHandlerMixin: print(f"Error sending metric: {e}") def send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool): - executor.submit(self._send_metric, start, end, success) + # executor.submit(self._send_metric, start, end, success) + self._send_metric(start, end, success) def poll(self): while True: From e0e7564fcc21c19654fbf48763ded277eb793214 Mon Sep 17 00:00:00 2001 From: Egor Matveev Date: Sun, 15 Jun 2025 05:27:31 +0300 Subject: [PATCH 3/8] fix --- utils/queues.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/utils/queues.py b/utils/queues.py index a571358..a69857d 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -13,7 +13,7 @@ else: QUEUES_URL = 'http://queues:1239' -executor = ThreadPoolExecutor(max_workers=1) +executor = ThreadPoolExecutor(max_workers=4) class QueuesException(Exception): @@ -40,8 +40,8 @@ class TasksHandlerMixin: print(f"Error sending metric: {e}") def send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool): - # executor.submit(self._send_metric, start, end, success) - self._send_metric(start, end, success) + executor.submit(self._send_metric, start, end, success) + # self._send_metric(start, end, success) def poll(self): while True: From 72c5823ccd528d91c9159a4be7df375ffb3e5ffa Mon Sep 17 00:00:00 2001 From: Egor Matveev Date: Sun, 15 Jun 2025 14:40:14 +0300 Subject: [PATCH 4/8] fix --- utils/queues.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/utils/queues.py b/utils/queues.py index a69857d..035b884 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -13,14 +13,15 @@ else: QUEUES_URL = 'http://queues:1239' -executor = ThreadPoolExecutor(max_workers=4) - - class QueuesException(Exception): ... class TasksHandlerMixin: + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.executor = ThreadPoolExecutor(max_workers=4) + def _send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool): try: resp = requests.post('http://monitoring:1237/api/v1/metrics/task', json={ @@ -39,10 +40,6 @@ class TasksHandlerMixin: except Exception as e: print(f"Error sending metric: {e}") - def send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool): - executor.submit(self._send_metric, start, end, success) - # self._send_metric(start, end, success) - def poll(self): while True: try: @@ -70,7 +67,7 @@ class TasksHandlerMixin: raise QueuesException except: print(f'Failed to finish task id={task["id"]}') - self.send_metric(start, end, success) + self.executor.submit(self._send_metric, start, end, success) @property def queue_name(self): From 2d292dfc46f46f5827c03222cffa55c94dcf2e91 Mon Sep 17 00:00:00 2001 From: Egor Matveev Date: Sun, 15 Jun 2025 14:42:46 +0300 Subject: [PATCH 5/8] fix --- utils/queues.py | 1 + 1 file changed, 1 insertion(+) diff --git a/utils/queues.py b/utils/queues.py index 035b884..9b77c8e 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -54,6 +54,7 @@ class TasksHandlerMixin: continue start = datetime.datetime.now(zoneinfo.ZoneInfo("Europe/Moscow")) try: + print(f'process task with id {task["id"]}') self.process(task['payload']) success = True except Exception as exc: From 0cab3e52cb167f18cdf03cb1fd8a3b0fd6e3122b Mon Sep 17 00:00:00 2001 From: Egor Matveev Date: Sun, 15 Jun 2025 14:45:14 +0300 Subject: [PATCH 6/8] fix --- utils/queues.py | 1 + 1 file changed, 1 insertion(+) diff --git a/utils/queues.py b/utils/queues.py index 9b77c8e..12b38df 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -66,6 +66,7 @@ class TasksHandlerMixin: 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.executor.submit(self._send_metric, start, end, success) From 4f0114e99ad812a95991103dba0f55bb7dacce26 Mon Sep 17 00:00:00 2001 From: Egor Matveev Date: Sun, 15 Jun 2025 14:53:13 +0300 Subject: [PATCH 7/8] fix --- utils/queues.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/utils/queues.py b/utils/queues.py index 12b38df..79559e1 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -54,7 +54,7 @@ class TasksHandlerMixin: continue start = datetime.datetime.now(zoneinfo.ZoneInfo("Europe/Moscow")) try: - print(f'process task with id {task["id"]}') + print(f'process task with id {task["id"]}, attempt {task["attempt"]}') self.process(task['payload']) success = True except Exception as exc: From d61c665b6c24509dbccdf302aaf8218ad359ef33 Mon Sep 17 00:00:00 2001 From: Egor Matveev Date: Sun, 15 Jun 2025 16:00:22 +0300 Subject: [PATCH 8/8] fix --- Dockerfile | 4 +++- daemons/metrics.py | 22 ++++++++++++++++++++++ main.py | 3 +++ run.sh | 13 +++++++++++++ utils/queues.py | 21 ++++++--------------- 5 files changed, 47 insertions(+), 16 deletions(-) create mode 100644 daemons/metrics.py create mode 100644 run.sh 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 79559e1..5adf3f4 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -1,6 +1,7 @@ -from concurrent.futures import ThreadPoolExecutor import datetime +import json import os +import uuid import zoneinfo import requests import time @@ -18,13 +19,9 @@ class QueuesException(Exception): class TasksHandlerMixin: - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.executor = ThreadPoolExecutor(max_workers=4) - def _send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool): - try: - resp = requests.post('http://monitoring:1237/api/v1/metrics/task', json={ + with open(f'/usr/src/{uuid.uuid4()}.json', 'w') as fp: + fp.write(json.dumps({ 'service': 'botalka', 'queue': self.queue_name, 'success': success, @@ -32,13 +29,7 @@ class TasksHandlerMixin: "success": success, "execution_time_ms": (end - start).microseconds // 1000, "environment": stage, - }) - if resp.status_code == 202: - print("Metric ok") - else: - print(f'metric not ok: {resp.status_code}') - except Exception as e: - print(f"Error sending metric: {e}") + })) def poll(self): while True: @@ -69,7 +60,7 @@ class TasksHandlerMixin: print(f'finish task with id {task["id"]}') except: print(f'Failed to finish task id={task["id"]}') - self.executor.submit(self._send_metric, start, end, success) + self._send_metric(start, end, success) @property def queue_name(self):