252 lines
20 KiB
YAML
252 lines
20 KiB
YAML
---
|
||
apiVersion: apps/v1
|
||
kind: Deployment
|
||
metadata:
|
||
name: worker-celery-beat
|
||
namespace: flows
|
||
labels:
|
||
app: worker-celery-beat
|
||
spec:
|
||
progressDeadlineSeconds: 600
|
||
replicas: 1
|
||
revisionHistoryLimit: 10
|
||
selector:
|
||
matchLabels:
|
||
app: worker-celery-beat
|
||
strategy:
|
||
rollingUpdate:
|
||
maxSurge: 25%
|
||
maxUnavailable: 25%
|
||
type: RollingUpdate
|
||
template:
|
||
metadata:
|
||
labels:
|
||
app: worker-celery-beat
|
||
spec:
|
||
affinity:
|
||
podAntiAffinity:
|
||
requiredDuringSchedulingIgnoredDuringExecution:
|
||
- labelSelector:
|
||
matchExpressions:
|
||
- key: app
|
||
operator: In
|
||
values:
|
||
- worker-celery-beat
|
||
topologyKey: kubernetes.io/hostname
|
||
containers:
|
||
- command:
|
||
- celery
|
||
- -A
|
||
- src.worker
|
||
- beat
|
||
- -l
|
||
- DEBUG
|
||
env:
|
||
- name: WORKFLOWS_NOTIFICATIONS_FROM_EMAIL
|
||
value: sarex@dsinv.ru
|
||
- name: ENABLE_METRICS
|
||
value: '0'
|
||
- name: ENABLE_MAILGUN
|
||
value: '0'
|
||
- name: SMTP_HOST
|
||
value: relay.dsinv.ru
|
||
- name: SMTP_PORT
|
||
value: '25'
|
||
- name: FROM_EMAIL
|
||
value: sarex@dsinv.ru
|
||
- name: MAILGUN_HOST
|
||
value: http:localhost:8000
|
||
- name: MAILGUN_API_KEY
|
||
value: empty
|
||
- name: NOTIFICATION_SETTINGS_USE_MAILGUN
|
||
value: '0'
|
||
- name: WORKFLOWS_HOST
|
||
value: http://workflows-service.workflow.svc.cluster.local:8000/api/v1
|
||
- name: FLOWS_HOST
|
||
value: http://backend-service.flows.svc.cluster.local:8000
|
||
- name: WORKFLOWS_NOTIFICATIONS_REGISTRY
|
||
value: cr.yandex/crp3ccidau046kdj8g9q
|
||
- name: WORKFLOWS_NOTIFICATIONS_SMTP_HOST
|
||
value: relay.dsinv.ru
|
||
- name: WORKFLOWS_NOTIFICATIONS_SMTP_PORT
|
||
value: '25'
|
||
- name: PLANNING_HOST
|
||
value: http://backend-service.pm.svc.cluster.local:8000/api/pm/msp
|
||
- name: PLANNING_USE
|
||
value: 'True'
|
||
- name: WORKFLOWS_NOTIFICATIONS_FROM_EMAI
|
||
value: sarex@dsinv.ru
|
||
- name: FLOWS_DB_HOST
|
||
value: postgres-service.flows.svc.cluster.local
|
||
- name: FLOWS_DB_DB
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: database
|
||
name: postgres-secret
|
||
- name: FLOWS_DB_USERNAME
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: username
|
||
name: postgres-secret
|
||
- name: FLOWS_DB_PASSWORD
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: password
|
||
name: postgres-secret
|
||
- name: ISSUES_DB_HOST
|
||
value: postgres-service.issues.svc.cluster.local
|
||
- name: ISSUES_DB_DB
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: database
|
||
name: postgres-secret-issues
|
||
- name: ISSUES_DB_USERNAME
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: username
|
||
name: postgres-secret-issues
|
||
- name: ISSUES_DB_PASSWORD
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: password
|
||
name: postgres-secret-issues
|
||
- name: DEBUG
|
||
value: '0'
|
||
- name: PG_PORT
|
||
value: '5432'
|
||
- name: FLOWS_DB_PORT
|
||
value: '5432'
|
||
- name: ISSUES_DB_PORT
|
||
value: '5432'
|
||
- name: DJANGO_HOST
|
||
value: http://backend.django.svc.cluster.local:8000/api
|
||
- name: DJANGO_BASE_HOST
|
||
value: https://sarex.dsinv.ru
|
||
- name: DJANGO_AUTH
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: token
|
||
name: django-secret
|
||
- name: DOCUMENTATION_HOST
|
||
value: http://documentations-service.documentations.svc.cluster.local:8080/internal/v1
|
||
- name: BASE_HOST
|
||
value: https://sarex.dsinv.ru
|
||
- name: RESOURCES_HOST
|
||
value: http://resources-service.resources.svc.cluster.local:8000/
|
||
- name: TIMEOUT
|
||
value: '120'
|
||
- name: RESOURCE_URL
|
||
value: http://resources-service.resources/api/v1
|
||
- name: GATEWAY_URL
|
||
value: http://pdm-api.documentations.svc.cluster.local:8080/api/v1
|
||
- name: SYNC_RESOURCE_ID
|
||
value: '1'
|
||
- name: SERVICE_HOST
|
||
value: https://sarex.dsinv.ru/flows/api/v1
|
||
- name: ENABLE_ANALYTICS
|
||
value: '1'
|
||
- name: ENABLE_CELERY
|
||
value: '1'
|
||
- name: CELERY_QUEUE
|
||
value: flow
|
||
- name: RABBITMQ_HOST
|
||
value: rabbitmq-service.flows.svc
|
||
- name: RABBITMQ_PORT
|
||
value: '5672'
|
||
- name: RABBITMQ_VHOST
|
||
value: flow
|
||
- name: PG_HOST
|
||
value: postgres-service.flows.svc.cluster.local
|
||
- name: PG_DB
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: database
|
||
name: postgres-secret
|
||
- name: PG_LOGIN
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: username
|
||
name: postgres-secret
|
||
- name: PG_PASSWORD
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: password
|
||
name: postgres-secret
|
||
- name: DJANGO_TOKEN
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: token
|
||
name: django-secret
|
||
- name: RABBITMQ_USERNAME
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: username
|
||
name: rabbitmq-secret
|
||
- name: RABBITMQ_PASSWORD
|
||
valueFrom:
|
||
secretKeyRef:
|
||
key: password
|
||
name: rabbitmq-secret
|
||
image: cr.yandex/crp3ccidau046kdj8g9q/flows-backend_worke:dsinv_700a79e5
|
||
imagePullPolicy: IfNotPresent
|
||
name: worker-celery-beat
|
||
ports:
|
||
- containerPort: 8000
|
||
name: http
|
||
protocol: TCP
|
||
resources:
|
||
requests:
|
||
cpu: '1'
|
||
memory: 512Mi
|
||
terminationMessagePath: /dev/termination-log
|
||
terminationMessagePolicy: File
|
||
volumeMounts:
|
||
- mountPath: /opt/src/worker/notifications_task.py
|
||
name: notifications-task
|
||
subPath: notifications_task.py
|
||
- mountPath: /opt/src/worker/templates/daily_digest.html
|
||
name: daily-digest
|
||
subPath: daily_digest.html
|
||
dnsPolicy: ClusterFirst
|
||
imagePullSecrets:
|
||
- name: dockerhub
|
||
nodeSelector:
|
||
name: generic
|
||
restartPolicy: Always
|
||
schedulerName: default-scheduler
|
||
securityContext: {}
|
||
terminationGracePeriodSeconds: 30
|
||
volumes:
|
||
- configMap:
|
||
defaultMode: 420
|
||
items:
|
||
- key: notifications_task.py
|
||
path: notifications_task.py
|
||
name: notifications-task
|
||
name: notifications-task
|
||
- configMap:
|
||
defaultMode: 420
|
||
items:
|
||
- key: daily_digest.html
|
||
path: daily_digest.html
|
||
name: daily-digest
|
||
name: daily-digest
|
||
---
|
||
apiVersion: v1
|
||
data:
|
||
daily_digest.html: "{% extends \"base_email.html\" %}\n\n{%- macro expiration_date(date, today) -%}\n{%- if date -%}\n<span class=\"expiration {{ 'expired' if date.date() < today else ('critical' if date.date() == today else 'normal') }}\">{{\n date.strftime('%d.%m.%Y')\n}}</span>\n{%- else -%}\nБез срока\n{%- endif -%}\n{%- endmacro %}\n\n{%- macro review_link(review, base_host) -%}\n<a href=\"{{ base_host }}/reviews/{{ review['review_id'] }}\">{{ review['review_name'] | e }}</a>\n{%- endmacro %}\n\n{%- block content %}\n<p>На сегодня в Sarex вашей реакции ждут следующие согласования:</p>\n\n{% if force_headers or len(expired_reviews) + len(ongoing_reviews) > 0 -%}\n<p>Согласования ({{ ongoing_reviews | length + expired_reviews | length }}):</p>\n{%- endif %}\n\n{% if force_headers or len(expired_reviews) > 0 -%}\n<p class=\"subitem\">Просроченные задачи ({{ expired_reviews | length }}):</p>\n<ol>\n {% for review in expired_reviews -%}\n <li><p>\n Проект: {{ review['resource_name'] | e }}, Согласование: {{ review_link(review, base_host) }},<br>\n Шаг: {{ review['step_name'] | e }}, Срок задачи: {{ expiration_date(review['expires_at'], today) }}, ID: (# {{ review['review_id'] }} )\n </p></li>\n {%- endfor %}\n</ol>\n{%- endif %}\n\n{% if force_headers or len(ongoing_reviews) > 0 -%}\n<p class=\"subitem\">Активные задачи ({{ ongoing_reviews | length }}):</p>\n<ol>\n {% for review in ongoing_reviews -%}\n <li><p>\n Проект: {{ review['resource_name'] | e }}, Согласование: {{ review_link(review, base_host) }},<br>\n Шаг: {{ review['step_name'] | e }}, Срок задачи: {{ expiration_date(review['expires_at'], today) }}, ID: (# {{ review['review_id'] }} )\n </p></li>\n {%- endfor %}\n</ol>\n{%- endif %}\n\n<p>С уважением, <br>команда Sarex</p>\n{% endblock %}\n"
|
||
kind: ConfigMap
|
||
metadata:
|
||
name: daily-digest
|
||
namespace: flows
|
||
---
|
||
apiVersion: v1
|
||
data:
|
||
notifications_task.py: "from collections import defaultdict\nfrom datetime import datetime\nfrom pathlib import Path\nfrom zoneinfo import ZoneInfo\n\nfrom jinja2 import Environment, FileSystemLoader\n\nfrom src.worker.celery import celery_app\nfrom src.worker.notifications_config import (\n DjangoClient,\n FlowsDatabase,\n ResourcesClient,\n IssuesDatabase,\n MailgunClient,\n NotificationGlobalSettings,\n WorkflowsNotificationsClient,\n)\n\n\n@celery_app.task(name=\"notify_users\")\ndef notify_users():\n django_client = DjangoClient()\n resources_client = ResourcesClient()\n flows_db = FlowsDatabase()\n flows_db_connection = flows_db.get_connection()\n issues_db = IssuesDatabase()\n issues_db_connection = issues_db.get_connection()\n\n users = django_client.get_users_with_enabled_notifications()\n users = {user[\"id\"]: user for user in users}\n if not users:\n flows_db_connection.close()\n issues_db_connection.close()\n return\n resources = resources_client.get_resources()\n resources = {resource[\"id\"]: resource[\"name\"] for resource in resources}\n\n with flows_db_connection.cursor() as cursor:\n cursor.execute(\n f\"\"\"\n SELECT\n r.id AS review_id,\n tq.reviewer_id AS user_id,\n r.resource_id AS resource_id,\n r.name AS review_name,\n s.name AS step_name,\n tq.end_date AS expires_at\n FROM task_queue tq\n JOIN review r\n ON r.id = tq.review_id\n JOIN step s\n ON s.id = r.current_step_id\n WHERE tq.reviewer_id IN ({\",\".join(str(x) for x in users.keys())})\n AND tq.is_active\n \"\"\"\n )\n result = cursor.fetchall()\n flows_db_connection.close()\n\n reviews_tasks = defaultdict(list)\n for review_task in result:\n review_task[\"resource_name\"] = resources.get(str(review_task[\"resource_id\"]), \"Без проекта\")\n reviews_tasks[review_task[\"user_id\"]].append(review_task)\n\n with issues_db_connection.cursor() as cursor:\n cursor.execute(\n f\"\"\"\n SELECT\n i.public_id AS issue_id,\n i.author_id AS user_id,\n i.resource_id AS resource_id,\n i.title AS name,\n i.completion_date AS expires_at\n FROM issues_issue i\n JOIN issues_issuestatus s\n ON s.id = i.status_id\n WHERE author_id IN ({\",\".join(str(x) for x in users.keys())})\n AND s.name NOT IN ('created', 'done')\n AND i.deleted_at IS NULL\n \"\"\"\n )\n authors_result = cursor.fetchall()\n cursor.execute(\n f\"\"\"\n SELECT\n i.public_id AS issue_id,\n ru.responsible_id AS user_id,\n i.resource_id AS resource_id,\n i.title AS name,\n i.completion_date AS expires_at\n FROM issues_issue i\n JOIN issues_responsibleuser ru\n ON ru.issue_id_id = i.public_id\n JOIN issues_issuestatus s\n ON s.id = i.status_id\n WHERE ru.responsible_id IN ({\",\".join(str(x) for x in users.keys())})\n AND s.name = 'created'\n AND i.deleted_at IS NULL\n \"\"\"\n )\n responsibles_result = cursor.fetchall()\n cursor.execute(\n \"\"\"\n SELECT\n i.public_id AS issue_id,\n i.responsible_entities AS responsible_entities,\n i.resource_id AS resource_id,\n i.title AS name,\n i.completion_date AS expires_at\n FROM issues_issue i\n JOIN issues_issuestatus s\n ON s.id = i.status_id\n WHERE i.responsible_entities IS NOT NULL\n AND s.name = 'created'\n AND\
|
||
\ i.deleted_at IS NULL\n \"\"\"\n )\n responsibles_sa_result = cursor.fetchall()\n issues_db_connection.close()\n\n issues_tasks = defaultdict(list)\n for issue_task in authors_result + responsibles_result:\n issue_task[\"resource_name\"] = resources.get(str(issue_task[\"resource_id\"]), \"Без проекта\")\n issues_tasks[issue_task[\"user_id\"]].append(issue_task)\n\n for issue in responsibles_sa_result:\n responsible_entities = issue[\"responsible_entities\"]\n positions = set()\n departments = set()\n for entity in responsible_entities[\"entities_list\"]:\n if entity[\"model_name\"] == \"position\":\n positions.add(str(entity[\"service_account_id\"]))\n elif entity[\"model_name\"] == \"department\":\n departments.add(str(entity[\"service_account_id\"]))\n if responsible_entities[\"operation\"] == \"AND\":\n if positions and departments:\n condition = lambda user_positions, user_departments: \\\n positions & user_positions and departments & user_departments\n elif positions:\n condition = lambda user_positions, _: \\\n positions & user_positions\n else:\n condition = lambda _, user_departments: \\\n departments & user_departments\n elif responsible_entities[\"operation\"] == \"OR\":\n condition = lambda user_positions, user_departments: \\\n positions & user_positions or departments & user_departments\n else:\n condition = lambda *_: False\n\n for user in users.values():\n user_positions, user_departments = set([p[\"service_account_id\"] for p in user[\"positions\"]]), \\\n set([d[\"service_account_id\"] for d in user[\"departments\"]])\n if condition(user_positions, user_departments):\n issue[\"resource_name\"] = resources.get(str(issue[\"resource_id\"]), \"Без проекта\")\n issues_tasks[user[\"id\"]].append(issue)\n\n jinja_env = Environment(loader=FileSystemLoader(Path(__file__).parent / \"templates\"))\n digest_template = jinja_env.get_template(\"daily_digest.html\")\n today = datetime.now(tz=ZoneInfo('UTC')).date()\n\n if NotificationGlobalSettings().use_mailgun:\n mailer = MailgunClient()\n else:\n mailer = WorkflowsNotificationsClient(job_name=\"Рассылка уведомлений: ежедневный дайджест\")\n\n for user_id, user in users.items():\n if not user[\"email\"]:\n continue\n reviews = reviews_tasks.get(user_id, [])\n issues = issues_tasks.get(user_id, [])\n if not (reviews or issues):\n continue\n\n expired_reviews = []\n ongoing_reviews = []\n for review in reviews:\n if bool(review[\"expires_at\"]) and review[\"expires_at\"].date() < today:\n expired_reviews.append(review)\n else:\n ongoing_reviews.append(review)\n\n # приклеиваем таймзону, чтобы все даты были offset-aware\n sorting_placeholder = datetime.max.replace(tzinfo=ZoneInfo('UTC'))\n expired_reviews.sort(key=lambda review: review[\"expires_at\"] or sorting_placeholder)\n ongoing_reviews.sort(key=lambda review: review[\"expires_at\"] or sorting_placeholder)\n issues.sort(key=lambda issue: issue[\"expires_at\"] or sorting_placeholder)\n\n html = digest_template.render(\n ongoing_reviews=ongoing_reviews,\n expired_reviews=expired_reviews,\n issues=issues,\n base_host=django_client.base_host,\n today=today,\n force_headers=True, # выводить ли хедеры, если сущностей ноль\n )\n\n mailer.send_email(\n send_to=user[\"email\"],\n subject=f\"Статус согласований в Sarex — {datetime.now(ZoneInfo('Europe/Moscow')):%d.%m.%Y}\",\n html=html,\n company_id=user[\"companies\"\
|
||
][0],\n triggerer_user_id=user_id,\n )\n\n\n@celery_app.task(name=\"notify_admins_about_empty_steps\")\ndef notify_admins_about_empty_steps():\n django_client = DjangoClient()\n resources_client = ResourcesClient()\n flows_db = FlowsDatabase()\n flows_db_connection = flows_db.get_connection()\n \n if NotificationGlobalSettings().use_mailgun:\n mailer = MailgunClient()\n else:\n mailer = WorkflowsNotificationsClient(job_name=\"Рассылка уведомлений: отсутствуют согласующие\")\n\n\n admins = django_client.get_flow_admins()\n admins = {user[\"id\"]: user for user in admins}\n\n if not admins:\n flows_db_connection.close()\n return\n\n resources = resources_client.get_resources()\n resources = {resource[\"id\"]: resource[\"name\"] for resource in resources}\n\n resource2users = resources_client.get_users_by_resources()\n resource2admins: dict[str, list[dict]] = defaultdict(list)\n for resource, user_ids in resource2users.items():\n for user_id in user_ids:\n if user_id in admins:\n resource2admins[resource].append(admins[user_id])\n\n with flows_db_connection.cursor() as cursor:\n cursor.execute(\n \"\"\"\n SELECT\n r.id AS review_id,\n r.resource_id AS resource_id,\n r.name AS review_name,\n f.company_id AS company_id,\n s.name AS step_name\n FROM review r\n JOIN flow f\n ON f.id = r.flow_id\n JOIN step s\n ON s.id = r.current_step_id\n WHERE r.status = 'open'\n AND r.current_reviewers = '{}'\n AND f.is_active\n \"\"\"\n )\n result = cursor.fetchall()\n flows_db_connection.close()\n\n jinja_env = Environment(loader=FileSystemLoader(Path(__file__).parent / \"templates\"))\n email_template = jinja_env.get_template(\"daily_empty_step.html\")\n\n for review in result:\n render_values = {\n 'review_name': review['review_name'],\n 'resource_name': resources.get(str(review[\"resource_id\"]), \"Без проекта\"),\n 'review_id': review['review_id'],\n 'current_step': review['step_name'],\n 'link': f'{django_client.base_host}/reviews/{review[\"review_id\"]}'\n }\n\n body = email_template.render(render_values)\n to = [user['email'] for user in resource2admins[str(review['resource_id'])]]\n subject = (\n f\"Напоминание: Отсутствуют согласующие — {render_values['resource_name']}\"\n f\" — {render_values['review_id']}\"\n f\" — {render_values['current_step']}\"\n f\" — {render_values['review_name']}\"\n )\n\n if len(to) > 0:\n user_id = resource2admins[str(review['resource_id'])][0]['id']\n\n mailer.send_email(\n send_to=','.join(to),\n subject=subject,\n html=body,\n company_id=review['company_id'],\n triggerer_user_id=user_id,\n )\n"
|
||
kind: ConfigMap
|
||
metadata:
|
||
name: notifications-task
|
||
namespace: flows
|