Compare commits

..

13 Commits

Author SHA1 Message Date
0e43995dfb Merge pull request 'fix' (#44) from master into prod
Reviewed-on: #44
2024-12-31 18:24:34 +03:00
f1169f3629 Merge pull request 'master' (#42) from master into prod
Reviewed-on: #42
2024-12-27 12:35:08 +03:00
b44ec9a1a4 Merge pull request 'Update daemons/worker.py' (#36) from master into prod
Reviewed-on: #36
2024-12-03 14:10:16 +03:00
1ac588785f Merge pull request 'Update daemons/worker.py' (#35) from master into prod
Reviewed-on: #35
2024-12-03 14:06:09 +03:00
d508e643bb Merge pull request 'fix' (#33) from master into prod
Reviewed-on: #33
2024-11-27 18:48:59 +03:00
62e61d21bf Merge pull request 'master' (#32) from master into prod
Reviewed-on: #32
2024-11-27 18:46:21 +03:00
5fd8147ff3 Merge pull request 'master' (#28) from master into prod
Reviewed-on: #28
2024-11-27 02:41:44 +03:00
adc2525fab Merge pull request 'master' (#23) from master into prod
Reviewed-on: #23
2024-11-24 23:50:29 +03:00
5967bc75d9 Merge pull request 'master' (#19) from master into prod
Reviewed-on: #19
2024-11-22 01:38:09 +03:00
f55b5f4841 Merge pull request 'queues' (#15) from queues into prod
Reviewed-on: #15
2024-11-17 13:32:22 +03:00
7713d24b63 Merge pull request 'fix' (#8) from master into prod
Reviewed-on: #8
2024-10-12 22:55:52 +03:00
1c82ef3b2b Merge pull request 'master' (#7) from master into prod
Reviewed-on: #7
2024-10-11 06:51:52 +03:00
a76656716c Merge pull request 'master' (#4) from master into prod
Reviewed-on: #4
2024-10-11 06:17:24 +03:00
4 changed files with 12 additions and 53 deletions

BIN
.DS_Store vendored

Binary file not shown.

View File

@@ -6,13 +6,12 @@ services:
image: mathwave/sprint-repo:pizda-bot
command: worker
environment:
MONGO_HOST: "mongo"
MONGO_HOST: "mongo.develop.sprinthub.ru"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
STAGE: "development"
networks:
- queues-development
- configurator
- mongo-development
deploy:
mode: replicated
restart_policy:
@@ -25,12 +24,11 @@ services:
image: mathwave/sprint-repo:pizda-bot
command: api
environment:
MONGO_HOST: "mongo"
MONGO_HOST: "mongo.develop.sprinthub.ru"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
STAGE: "development"
networks:
- common-infra-nginx-development
- mongo-development
deploy:
mode: replicated
restart_policy:
@@ -46,5 +44,3 @@ networks:
external: true
configurator:
external: true
mongo-development:
external: true

View File

@@ -6,13 +6,12 @@ services:
image: mathwave/sprint-repo:pizda-bot
command: worker
environment:
MONGO_HOST: "mongo"
MONGO_HOST: "mongo.sprinthub.ru"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
STAGE: "production"
networks:
- queues
- configurator
- mongo
deploy:
mode: replicated
restart_policy:
@@ -25,11 +24,10 @@ services:
image: mathwave/sprint-repo:pizda-bot
command: api
environment:
MONGO_HOST: "mongo"
MONGO_HOST: "mongo.sprinthub.ru"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
networks:
- common-infra-nginx
- mongo
deploy:
mode: replicated
restart_policy:
@@ -45,5 +43,3 @@ networks:
external: true
configurator:
external: true
mongo:
external: true

View File

@@ -1,7 +1,4 @@
from concurrent.futures import ThreadPoolExecutor
import datetime
import os
import zoneinfo
import requests
import time
@@ -18,54 +15,24 @@ class QueuesException(Exception):
class TasksHandlerMixin:
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.executor = ThreadPoolExecutor(max_workers=1)
def _send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool):
def send():
requests.post(f'{QUEUES_URL}/api/v1/metric', 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,
})
self.executor.submit(send)
def poll(self):
while True:
try:
response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name}).json()
except requests.JSONDecodeError:
print('Unable to decode json')
time.sleep(3)
continue
task = response.get('task')
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}')
success = False
end = datetime.datetime.now(zoneinfo.ZoneInfo("Europe/Moscow"))
if success:
continue
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):