Обработка оповещений Prometheus с помощью Fastapi
Немного начинающего вопроса: Я пытаюсь создать маршрут FastAPI, который обрабатывает предупреждения Prometheus. Само оповещение — это просто полезная нагрузка, которая выглядит так:
{
"endsAt": "0001-01-01T00:00:00Z",
"labels.prometheus": "label/prometheus-stack",
"labels.namespace": "namespace_here",
"labels.alertname": "JobFailed",
"labels.severity": "critical",
"labels.job_name": "uuid-job",
"labels.purpose": "purpose_here",
"labels.team": "team_here",
"status": "firing",
"startsAt": "2023-03-13T21:05:48.433Z",
"annotations.summary": "job failed",
"annotations.message": "Job XY failed.",
"fingerprint": "1234123412341234",
"generatorURL": "https://URL.stuff"
}
Для API у меня есть класс pydantic "Alert", который выглядит так:
class Alert(BaseModel):
fingerprint: str
status: str
startsAt: datetime
endsAt: datetime
endsAt: str
annotations: Dict[str, str]
labels: Dict[str, str]
И все, что я хочу сделать, это иметь возможность отправлять полезную нагрузку Prometheus на следующий маршрут FastAPI и иметь доступ ко всем свойствам и что-то с этим делать.
@api.post("/alerts/jobs", tags=["Alerts"])
async def recieve_failed_job_alert(alert: Alert):
print(alert)
print(alert.fingerprint)
print(alert.annotations.summary)
print(alert.labels.alertname)
Если у меня есть упомянутый класс pydantic, FastAPI отвечает «422 Unprocessable Entity», поскольку обязательны поля «аннотации» и «метки». Сами по себе они не имеют никакого контента, только «дочерние поля» (?), такие как labels.namespace, annotations.message,... имеют контент.
Если я отредактирую класс pydantic, чтобы задать эти поля по умолчанию и сделать так, чтобы они больше не требовались, например:
annotations: Dict[str, str] = None
labels: Dict[str, str] = None
то почтовый запрос выполнен успешно, но значения нежелательны. Данная полезная нагрузка задается как
fingerprint='1234123412341234' status='firing' starts_at=datetime.datetime(2023, 3, 13, 21, 5, 48, 433000, tzinfo=datetime.timezone.utc) ends_at=datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc) generator_url='https://URL.stuff' annotations=None labels=None labels.purpose='purpose_here' labels.prometheus='label/prometheus-stack' labels.namespace='namespace_here' labels.severity='critical' annotations.summary='job failed' annotations.message='Job XY failed.' labels.alertname='JobFailed' labels.team='team_here' labels.job_name='uuid-job'
Обратите внимание, что "...аннотации=Нет меток=Нет..." Кроме того, вложенные метки и аннотации, по-видимому, не подлежат подписке.
print(alert.labels.namespace)
^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'namespace'
Я уверен, что просто недостаточно понимаю обработку таких вложенных полей. Но из-за этого я также не знаю, как помочь себе и заставить это работать. Любая помощь приветствуется!
С уважением
...но, согласно этим документам, оповещение выглядит не так.
Данные, поступающие от прометея, настроены, я не могу это изменить и сам не причастен к самому прометею. Я просто использую полезную нагрузку, которая предоставляется. Кроме того, я также ожидал бы, что это интерпретируется как строка, и поэтому я мог бы просто получить к ней доступ, как если бы я получил доступ к «статусу». работает как положено, но print(alert.status) выдаст firing
Ответы 1
Ваши последние комментарии показывают некоторую путаницу в синтаксисе Python. Если в ваших данных JSON есть поле с именем labels.prometheus, вы не можете запросить у своей модели alert.labels.prometheus; это предполагает, что ваш объект Alert имеет поле с именем labels (которое, в свою очередь, имеет поле с именем prometheus).
У вас плоская структура данных. Не существует иерархии, которая позволяла бы точечное выражение, которое вы пытаетесь использовать.
Вам нужно создать модель, которая сопоставляет эти имена полей с точками (labels.prometheus) с именами полей, совместимыми с синтаксисом Python (например, labels_prometheus). Мы можем сделать это, используя псевдонимы полей; например:
from datetime import datetime
from pydantic import BaseModel, Field
from fastapi import FastAPI
class Alert(BaseModel):
fingerprint: str
status: str
startsAt: datetime
endsAt: datetime
endsAt: str
labels_prometheus: str = Field(alias = "labels.prometheus")
labels_namespace: str = Field(alias = "labels.namespace")
labels_alertname: str = Field(alias = "labels.alertname")
labels_severity: str = Field(alias = "labels.severity")
labels_job_name: str = Field(alias = "labels.job_name")
labels_purpose: str = Field(alias = "labels.purpose")
labels_team: str = Field(alias = "labels.team")
annotations_summary: str = Field(alias = "annotations.summary")
annotations_message: str = Field(alias = "annotations.message")
api = FastAPI()
@api.post("/alerts/jobs", tags=["Alerts"])
async def recieve_failed_job_alert(alert: Alert):
print(alert)
Если мы доставим ваш пример документа JSON в этот API, мы увидим напечатанное на консоли:
fingerprint='1234123412341234' status='firing'
startsAt=datetime.datetime(2023, 3, 13, 21, 5, 48, 433000,
tzinfo=datetime.timezone.utc) endsAt='0001-01-01T00:00:00Z'
labels_prometheus='label/prometheus-stack'
labels_namespace='namespace_here'
labels_alertname='JobFailed' labels_severity='critical'
labels_job_name='uuid-job' labels_purpose='purpose_here'
labels_team='team_here' annotations_summary='job failed'
annotations_message='Job XY failed.'
Подробнее в документации Pydantic.
В качестве альтернативы, если вам нужна иерархическая структура данных, вы можете преобразовать входные данные JSON с помощью root_validator. Что-то вроде этого, может быть:
from datetime import datetime
from pydantic import BaseModel, Field, validator, root_validator
from fastapi import FastAPI
class Alert(BaseModel):
fingerprint: str
status: str
startsAt: datetime
endsAt: datetime
endsAt: str
labels: dict[str, str]
annotations: dict[str, str]
@root_validator(pre=True)
def split_dots(cls, values):
newvalues = {}
for k, v in values.items():
if "." in k:
head, tail = k.split(".")
newvalues.setdefault(head, {})[tail] = v
else:
newvalues[k] = v
return newvalues
api = FastAPI()
@api.post("/alerts/jobs", tags=["Alerts"])
async def recieve_failed_job_alert(alert: Alert):
print(alert)
Это будет печатать:
fingerprint='1234123412341234' status='firing'
startsAt=datetime.datetime(2023, 3, 13, 21, 5, 48,
433000, tzinfo=datetime.timezone.utc)
endsAt='0001-01-01T00:00:00Z' labels = {'prometheus':
'label/prometheus-stack', 'namespace':
'namespace_here', 'alertname': 'JobFailed',
'severity': 'critical', 'job_name': 'uuid-job',
'purpose': 'purpose_here', 'team': 'team_here'}
annotations = {'summary': 'job failed', 'message': 'Job
XY failed.'}
Это именно то, что мне нужно. Большое спасибо!! И да, мне еще многому предстоит научиться...
Я думаю, вы неправильно читаете свои входные данные: нет полей с названиями «метки» или «аннотации»; это плоский документ без вложенности. У вас есть поля с такими именами, как «labels.prometheus», и это обычное строковое поле, такое же, как «status».