Sergey Antropov
b90def35ed
- FastAPI приложение для отправки мониторинговых алертов в мессенджеры - Поддержка Telegram и MAX/VK - Интеграция с Grafana, Zabbix, AlertManager - Автоматическое создание тикетов в Jira - Управление группами мессенджеров через API - Декораторы для авторизации и скрытия эндпоинтов - Подробная документация в папке docs/ Автор: Сергей Антропов Сайт: https://devops.org.ru
331 lines
12 KiB
Python
331 lines
12 KiB
Python
"""
|
||
Утилиты для работы с Jira тикетами.
|
||
|
||
Автор: Сергей Антропов
|
||
Сайт: https://devops.org.ru
|
||
"""
|
||
import logging
|
||
from typing import Optional, Dict, Any
|
||
from jinja2 import Environment, FileSystemLoader
|
||
|
||
from app.models.alertmanager import PrometheusAlert
|
||
from app.models.grafana import GrafanaAlert
|
||
from app.models.zabbix import ZabbixAlert
|
||
from app.models.jira import JiraMapping
|
||
from app.core.jira_client import JiraClient
|
||
from app.core.jira_mapping import jira_mapping_manager
|
||
from app.core.config import get_settings
|
||
from app.core.utils import add_spaces_to_alert_name
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def create_jira_ticket_from_alert(
|
||
alert: Any,
|
||
source: str,
|
||
k8s_cluster: Optional[str] = None,
|
||
mapping: Optional[JiraMapping] = None
|
||
) -> Optional[str]:
|
||
"""
|
||
Создать Jira тикет на основе алерта.
|
||
|
||
Args:
|
||
alert: Данные алерта (PrometheusAlert, GrafanaAlert, ZabbixAlert).
|
||
source: Источник алерта (alertmanager, grafana, zabbix).
|
||
k8s_cluster: Имя Kubernetes кластера (опционально).
|
||
mapping: Маппинг для создания тикета (опционально).
|
||
|
||
Returns:
|
||
Ключ созданного тикета или None в случае ошибки.
|
||
"""
|
||
from app.core.config import get_settings
|
||
settings = get_settings()
|
||
|
||
if not settings.jira_enabled:
|
||
logger.debug("Jira отключен, тикет не создается")
|
||
return None
|
||
|
||
if not settings.jira_url or not settings.jira_email or not settings.jira_api_token:
|
||
logger.warning("Jira не настроен (отсутствуют URL, email или API token)")
|
||
return None
|
||
|
||
# Создаем клиент Jira
|
||
try:
|
||
jira_client = JiraClient(
|
||
url=settings.jira_url,
|
||
email=settings.jira_email,
|
||
api_token=settings.jira_api_token
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка создания Jira клиента: {e}")
|
||
return None
|
||
|
||
# Получаем маппинг, если не указан
|
||
if not mapping:
|
||
alert_data = _extract_alert_data(alert, source, k8s_cluster)
|
||
mapping = await jira_mapping_manager.find_mapping(source, alert_data)
|
||
|
||
if not mapping:
|
||
logger.warning(f"Маппинг не найден для источника {source}")
|
||
return None
|
||
|
||
# Формируем данные тикета
|
||
summary = _generate_jira_summary(alert, source, mapping)
|
||
description = _generate_jira_description(alert, source, k8s_cluster, mapping)
|
||
|
||
# Создаем тикет
|
||
try:
|
||
issue_key = jira_client.create_issue(
|
||
project=mapping.project,
|
||
summary=summary,
|
||
description=description,
|
||
issue_type=mapping.issue_type,
|
||
assignee=mapping.assignee,
|
||
priority=mapping.priority,
|
||
labels=mapping.labels
|
||
)
|
||
|
||
if issue_key:
|
||
logger.info(f"Jira тикет {issue_key} создан для алерта из {source}")
|
||
# Увеличиваем счетчик созданных тикетов
|
||
from app.core.metrics import metrics
|
||
metrics.increment_jira_ticket_created(
|
||
source=source,
|
||
project=mapping.project,
|
||
k8s_cluster=k8s_cluster or "",
|
||
chat="",
|
||
thread=0
|
||
)
|
||
|
||
return issue_key
|
||
except Exception as e:
|
||
logger.error(f"Ошибка создания Jira тикета: {e}")
|
||
# Увеличиваем счетчик ошибок
|
||
from app.core.metrics import metrics
|
||
metrics.increment_jira_ticket_error(
|
||
source=source,
|
||
k8s_cluster=k8s_cluster or "",
|
||
chat="",
|
||
thread=0
|
||
)
|
||
return None
|
||
|
||
|
||
def _extract_alert_data(
|
||
alert: Any,
|
||
source: str,
|
||
k8s_cluster: Optional[str] = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
Извлечь данные алерта для маппинга.
|
||
|
||
Args:
|
||
alert: Данные алерта.
|
||
source: Источник алерта.
|
||
k8s_cluster: Имя Kubernetes кластера.
|
||
|
||
Returns:
|
||
Словарь с данными алерта.
|
||
"""
|
||
alert_data = {}
|
||
|
||
if source == "alertmanager" and isinstance(alert, PrometheusAlert):
|
||
alert_data["status"] = alert.status
|
||
alert_data.update(alert.commonLabels)
|
||
alert_data.update(alert.commonAnnotations)
|
||
if k8s_cluster:
|
||
alert_data["k8s_cluster"] = k8s_cluster
|
||
elif source == "grafana" and isinstance(alert, GrafanaAlert):
|
||
alert_data["state"] = alert.state
|
||
alert_data["tags"] = alert.tags
|
||
alert_data["ruleName"] = alert.ruleName
|
||
elif source == "zabbix" and isinstance(alert, ZabbixAlert):
|
||
alert_data["status"] = alert.status
|
||
alert_data["event-severity"] = alert.event_severity or ""
|
||
alert_data["event-name"] = alert.event_name
|
||
alert_data["host-name"] = alert.host_name
|
||
|
||
return alert_data
|
||
|
||
|
||
def _generate_jira_summary(
|
||
alert: Any,
|
||
source: str,
|
||
mapping: JiraMapping
|
||
) -> str:
|
||
"""
|
||
Сгенерировать заголовок Jira тикета.
|
||
|
||
Args:
|
||
alert: Данные алерта.
|
||
source: Источник алерта.
|
||
mapping: Маппинг для создания тикета.
|
||
|
||
Returns:
|
||
Заголовок тикета.
|
||
"""
|
||
severity_prefix = ""
|
||
alert_name = ""
|
||
|
||
if source == "alertmanager" and isinstance(alert, PrometheusAlert):
|
||
severity = alert.commonLabels.get("severity", "")
|
||
alert_name = alert.commonLabels.get("alertname", "")
|
||
if severity:
|
||
severity_prefix = f"[{severity.upper()}] "
|
||
if alert_name:
|
||
alert_name = add_spaces_to_alert_name(alert_name)
|
||
summary = alert.commonAnnotations.get("summary", alert_name)
|
||
elif source == "grafana" and isinstance(alert, GrafanaAlert):
|
||
alert_name = alert.ruleName
|
||
if alert.state == "alerting":
|
||
severity_prefix = "[ALERTING] "
|
||
summary = alert.title or alert_name
|
||
elif source == "zabbix" and isinstance(alert, ZabbixAlert):
|
||
severity = alert.event_severity or ""
|
||
alert_name = alert.event_name
|
||
if severity:
|
||
severity_prefix = f"[{severity.upper()}] "
|
||
summary = alert.alert_subject or alert_name
|
||
else:
|
||
summary = "Unknown Alert"
|
||
|
||
return f"{severity_prefix}{alert_name}: {summary}"[:255] # Ограничение Jira
|
||
|
||
|
||
def _generate_jira_description(
|
||
alert: Any,
|
||
source: str,
|
||
k8s_cluster: Optional[str] = None,
|
||
mapping: Optional[JiraMapping] = None
|
||
) -> str:
|
||
"""
|
||
Сгенерировать описание Jira тикета.
|
||
|
||
Args:
|
||
alert: Данные алерта.
|
||
source: Источник алерта.
|
||
k8s_cluster: Имя Kubernetes кластера.
|
||
mapping: Маппинг для создания тикета.
|
||
|
||
Returns:
|
||
Описание тикета в формате Markdown.
|
||
"""
|
||
from app.core.config import get_settings
|
||
settings = get_settings()
|
||
|
||
# Загружаем шаблон описания
|
||
try:
|
||
environment = Environment(loader=FileSystemLoader(settings.templates_path))
|
||
template_name = f"jira_{source}.tmpl"
|
||
try:
|
||
template = environment.get_template(template_name)
|
||
except Exception:
|
||
# Если шаблон не найден, используем общий шаблон
|
||
template = environment.get_template("jira_common.tmpl")
|
||
except Exception:
|
||
# Если общий шаблон не найден, используем простое описание
|
||
return _generate_simple_description(alert, source, k8s_cluster)
|
||
|
||
# Формируем словарь для шаблона
|
||
template_data = _prepare_template_data(alert, source, k8s_cluster)
|
||
|
||
# Рендерим шаблон
|
||
description = template.render(template_data)
|
||
|
||
return description
|
||
|
||
|
||
def _generate_simple_description(
|
||
alert: Any,
|
||
source: str,
|
||
k8s_cluster: Optional[str] = None
|
||
) -> str:
|
||
"""
|
||
Сгенерировать простое описание тикета без шаблона.
|
||
|
||
Args:
|
||
alert: Данные алерта.
|
||
source: Источник алерта.
|
||
k8s_cluster: Имя Kubernetes кластера.
|
||
|
||
Returns:
|
||
Простое описание тикета.
|
||
"""
|
||
description = f"**Источник:** {source}\n\n"
|
||
|
||
if source == "alertmanager" and isinstance(alert, PrometheusAlert):
|
||
description += f"**Статус:** {alert.status}\n\n"
|
||
description += "**Метки:**\n"
|
||
for key, value in alert.commonLabels.items():
|
||
description += f"- {key}: {value}\n"
|
||
description += "\n**Аннотации:**\n"
|
||
for key, value in alert.commonAnnotations.items():
|
||
description += f"- {key}: {value}\n"
|
||
if k8s_cluster:
|
||
description += f"\n**Kubernetes кластер:** {k8s_cluster}\n"
|
||
elif source == "grafana" and isinstance(alert, GrafanaAlert):
|
||
description += f"**Состояние:** {alert.state}\n\n"
|
||
description += f"**Правило:** {alert.ruleName}\n\n"
|
||
description += f"**Сообщение:** {alert.message or 'Нет сообщения'}\n\n"
|
||
if alert.tags:
|
||
description += "**Теги:**\n"
|
||
for key, value in alert.tags.items():
|
||
description += f"- {key}: {value}\n"
|
||
elif source == "zabbix" and isinstance(alert, ZabbixAlert):
|
||
description += f"**Статус:** {alert.status}\n\n"
|
||
description += f"**Серьезность:** {alert.event_severity or 'Unknown'}\n\n"
|
||
description += f"**Событие:** {alert.event_name}\n\n"
|
||
description += f"**Хост:** {alert.host_name} ({alert.host_ip})\n\n"
|
||
description += f"**Сообщение:** {alert.alert_message}\n"
|
||
|
||
return description
|
||
|
||
|
||
def _prepare_template_data(
|
||
alert: Any,
|
||
source: str,
|
||
k8s_cluster: Optional[str] = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
Подготовить данные для шаблона описания тикета.
|
||
|
||
Args:
|
||
alert: Данные алерта.
|
||
source: Источник алерта.
|
||
k8s_cluster: Имя Kubernetes кластера.
|
||
|
||
Returns:
|
||
Словарь с данными для шаблона.
|
||
"""
|
||
template_data = {
|
||
"source": source,
|
||
"k8s_cluster": k8s_cluster or "",
|
||
}
|
||
|
||
if source == "alertmanager" and isinstance(alert, PrometheusAlert):
|
||
template_data["status"] = alert.status
|
||
template_data["common_labels"] = alert.commonLabels
|
||
template_data["common_annotations"] = alert.commonAnnotations
|
||
template_data["alertname"] = alert.commonLabels.get("alertname", "")
|
||
template_data["severity"] = alert.commonLabels.get("severity", "")
|
||
template_data["summary"] = alert.commonAnnotations.get("summary", "")
|
||
template_data["description"] = alert.commonAnnotations.get("description", "")
|
||
elif source == "grafana" and isinstance(alert, GrafanaAlert):
|
||
template_data["state"] = alert.state
|
||
template_data["title"] = alert.title
|
||
template_data["ruleName"] = alert.ruleName
|
||
template_data["message"] = alert.message or ""
|
||
template_data["tags"] = alert.tags
|
||
template_data["evalMatches"] = alert.evalMatches
|
||
elif source == "zabbix" and isinstance(alert, ZabbixAlert):
|
||
template_data["status"] = alert.status
|
||
template_data["event_severity"] = alert.event_severity or ""
|
||
template_data["event_name"] = alert.event_name
|
||
template_data["alert_subject"] = alert.alert_subject
|
||
template_data["alert_message"] = alert.alert_message
|
||
template_data["host_name"] = alert.host_name
|
||
template_data["host_ip"] = alert.host_ip
|
||
template_data["host_port"] = alert.host_port
|
||
|
||
return template_data
|