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/daemons/poll.py b/daemons/poll.py index d7169ee..d3ad377 100644 --- a/daemons/poll.py +++ b/daemons/poll.py @@ -1,6 +1,7 @@ import telebot import multiprocessing import time +import json from daemons import base from utils import platform diff --git a/utils/queues.py b/utils/queues.py index dc0f219..8cf6aaa 100644 --- a/utils/queues.py +++ b/utils/queues.py @@ -1,4 +1,6 @@ +import datetime import os +from threading import Thread import requests import time @@ -15,6 +17,27 @@ class QueuesException(Exception): class TasksHandlerMixin: + def _send_metric(self, start, success, end): + try: + metric = requests.post('http://monitoring:1237/api/v1/metrics/task', json={ + 'timestamp': start.strftime("%Y-%m-%dT%H:%M:%S") + "Z", + 'service': 'botalka', + 'environment': stage, + 'queue': self.queue_name, + 'success': success, + 'execution_time_ms': (end - start).microseconds // 1000, + }) + if metric.status_code == 202: + print('metric ok') + else: + print(f'metric failed: {metric.status_code}') + except Exception as e: + print(f'metric failed: {e}') + + def send_metric(self, start, success, end): + # Thread(target=self._send_metric, args=(start, success, end)).start() + self._send_metric(start, success, end) + def poll(self): while True: try: @@ -27,17 +50,21 @@ class TasksHandlerMixin: if not task: time.sleep(0.2) continue + start = datetime.datetime.now() 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 + success = False + end = datetime.datetime.now() 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, success, end) @property def queue_name(self):