monitoring.alerts_storage module

MongoDB-backed alerts storage with TTL and simple counters.

Design goals: - Fail-open: never throw from public APIs - Lazy init: connect to Mongo only on first use - Config via env only (avoid importing global config to prevent cycles) - TTL-based retention (default 30 days) to keep the collection bounded

Environment variables: - ALERTS_DB_ENABLED: ”true/1/yes“ to enable writes (fallback to METRICS_DB_ENABLED) - MONGODB_URL: required when enabled - DATABASE_NAME: DB name (default: code_keeper_bot) - ALERTS_COLLECTION: Collection name (default: alerts_log) - ALERTS_TTL_DAYS: TTL for documents (default: 30)

Public API: - record_alert(alert_id, name, severity, summary, source) -> None - count_alerts_since(since_dt) -> tuple[int, int] - count_alerts_last_hours(hours=24) -> tuple[int, int]

monitoring.alerts_storage.compute_error_signature(error_data)[מקור]

מחשב חתימה ייחודית לשגיאה.

החתימה מבוססת על: - סוג השגיאה - שם הקובץ והשורה (אם יש) - 3 השורות הראשונות של ה-stack trace

Return type:

str

פרמטרים:

error_data (Dict[str, Any])

monitoring.alerts_storage.is_new_error(signature)[מקור]

בודק אם השגיאה חדשה (לא נראתה ב-30 יום האחרונים).

Return type:

bool

פרמטרים:

signature (str)

monitoring.alerts_storage.enrich_alert_with_signature(alert_data)[מקור]

מעשיר את נתוני ההתראה עם חתימה ומידע על חדשות.

Return type:

Dict[str, Any]

פרמטרים:

alert_data (Dict[str, Any])

monitoring.alerts_storage.record_alert(*, alert_id, name, severity, summary='', source='', silenced=False, details=None)[מקור]

Insert (or upsert via unique key) a single alert record.

  • When alert_id is provided, use it for de-duplication via a unique key.

  • Otherwise use a stable hash based on name/severity/summary/minute.

Return type:

None

פרמטרים:
monitoring.alerts_storage.fetch_alert_type_catalog(*, min_total_count=1, limit=5000)[מקור]

Return catalog of all observed alert types (fail-open).

Return type:

List[Dict[str, Any]]

פרמטרים:

  • min_total_count (int)

  • limit (int)

Each row includes:

alert_type, total_count, first_seen_dt, last_seen_dt, sample_name, sample_title

monitoring.alerts_storage.fetch_alerts_by_type(*, alert_type, limit=100, include_details=True)[מקור]

Fetch recent alerts of a specific type with Sentry details.

Returns a list of dicts, for example:

{
  "alert_id": str,
  "ts_dt": datetime,
  "name": str,
  "summary": str,
  "sentry_issue_id": Optional[str],
  "sentry_permalink": Optional[str],
  "sentry_short_id": Optional[str],
}
Return type:

List[Dict[str, Any]]

פרמטרים:

  • alert_type (str)

  • limit (int)

  • include_details (bool)

monitoring.alerts_storage.count_alerts_since(since_dt)[מקור]

Return (total, critical) counts since the given datetime (UTC recommended).

Return type:

tuple[int, int]

פרמטרים:

since_dt (datetime)

monitoring.alerts_storage.count_alerts_last_hours(hours=24)[מקור]
Return type:

tuple[int, int]

פרמטרים:

hours (int)

monitoring.alerts_storage.list_recent_alert_ids(limit=10)[מקור]

Return recent alert identifiers from the DB (fail-open).

Preference order: document alert_id when present, otherwise the stable unique _key used for de-duplication. Results are ordered by ts_dt descending and truncated to limit.

Return type:

List[str]

פרמטרים:

limit (int)

monitoring.alerts_storage.fetch_alerts(*, start_dt, end_dt, severity=None, alert_type=None, endpoint=None, search=None, page=1, per_page=50)[מקור]

Return paginated alert documents filtered by the provided criteria.

Return type:

Tuple[List[Dict[str, Any]], int]

פרמטרים:
monitoring.alerts_storage.aggregate_alert_type_stats(*, start_dt, end_dt, min_count=1, limit=500)[מקור]

Aggregate active alert_types with counts and last_seen (exclude drills).

Return type:

List[Dict[str, Any]]

פרמטרים:
Returns a list of dicts:

{ ”alert_type“: str, ”count“: int, ”last_seen_dt“: datetime, ”sample_title“: str, ”sample_name“: str }

Fail-open: returns [] on any error / when storage is unavailable.

monitoring.alerts_storage.aggregate_alert_summary(*, start_dt, end_dt)[מקור]

Aggregate alert counts by severity and deployment flag.

Return type:

Dict[str, int]

פרמטרים:
monitoring.alerts_storage.fetch_alert_timestamps(*, start_dt, end_dt, severity=None, alert_type=None, limit=500)[מקור]

Return recent alert timestamps matching the given filters.

Return type:

List[datetime]

פרמטרים:
monitoring.alerts_storage.aggregate_alert_timeseries(*, start_dt, end_dt, granularity_seconds)[מקור]

Aggregate alert counts per severity over time buckets.

Return type:

List[Dict[str, Any]]

פרמטרים: