|
1 | 1 | """Prometheus metrics for the notification queue. |
2 | 2 |
|
3 | | -Exposes two gauges that, together, catch a stuck worker, a bad Slack token, |
| 3 | +Exposes gauges that, together, catch a stuck worker, a bad Slack token, |
4 | 4 | or a Slack outage: |
5 | 5 |
|
6 | 6 | - ``esb_pending_notifications_count`` — count of rows with status='pending'. |
7 | 7 | - ``esb_oldest_pending_notification_timestamp_seconds`` — Unix epoch seconds |
8 | 8 | of the oldest pending row's ``created_at``. **Omitted entirely** when the |
9 | 9 | table has no pending rows; Prometheus alert rules should use ``absent()`` |
10 | 10 | rather than a sentinel value. |
| 11 | +- ``esb_worker_last_iteration_timestamp_seconds`` — Unix epoch seconds of the |
| 12 | + worker's last successful poll cycle (read from ``AppConfig``). Omitted when |
| 13 | + the worker has never run or the underlying query fails. |
| 14 | +- ``esb_socket_mode_enabled`` / ``esb_socket_mode_connected`` — Slack Socket |
| 15 | + Mode intent and binding state. Always emitted. |
11 | 16 |
|
12 | 17 | Alert rules belong in Prometheus, not here. Example:: |
13 | 18 |
|
14 | 19 | time() - esb_oldest_pending_notification_timestamp_seconds > 300 |
15 | 20 | """ |
16 | 21 |
|
17 | | -from datetime import UTC |
| 22 | +import logging |
| 23 | +from datetime import UTC, datetime |
18 | 24 |
|
19 | 25 | from prometheus_client import CollectorRegistry, generate_latest |
20 | 26 | from prometheus_client.core import GaugeMetricFamily |
21 | 27 | from prometheus_client.exposition import CONTENT_TYPE_LATEST |
| 28 | +from sqlalchemy import select |
| 29 | +from sqlalchemy.exc import SQLAlchemyError |
22 | 30 |
|
23 | 31 | from esb.extensions import db |
| 32 | +from esb.models.app_config import AppConfig |
24 | 33 | from esb.models.pending_notification import PendingNotification |
25 | 34 |
|
| 35 | +logger = logging.getLogger(__name__) |
| 36 | + |
| 37 | +# Per-process latch: the first AppConfig query failure logs a full traceback |
| 38 | +# (diagnostically useful), subsequent failures log a one-line warning. Prevents |
| 39 | +# log flooding on pre-migration deployments where /metrics is scraped every |
| 40 | +# 15-30s and the app_config table doesn't yet exist. |
| 41 | +_app_config_query_failed_once: bool = False |
| 42 | + |
26 | 43 |
|
27 | 44 | def _query_pending_stats() -> tuple[int, float | None]: |
28 | 45 | """Return (pending_count, oldest_created_at_unix_seconds_or_None). |
@@ -81,12 +98,109 @@ def collect(self): |
81 | 98 | ) |
82 | 99 |
|
83 | 100 |
|
| 101 | +class _WorkerStatusCollector: |
| 102 | + """Custom collector for worker-liveness and Slack Socket Mode state. |
| 103 | +
|
| 104 | + Emits up to three gauges per scrape: |
| 105 | + - ``esb_worker_last_iteration_timestamp_seconds`` (omitted if absent or |
| 106 | + the AppConfig query fails) |
| 107 | + - ``esb_socket_mode_enabled`` (always emitted) |
| 108 | + - ``esb_socket_mode_connected`` (always emitted) |
| 109 | +
|
| 110 | + Failures of the AppConfig query never propagate out of ``collect()`` — the |
| 111 | + worker-timestamp gauge is simply omitted. The Socket Mode gauges still |
| 112 | + emit so the actionable failure mode (enabled=1, connected=0) remains |
| 113 | + observable even when the DB is unavailable. |
| 114 | + """ |
| 115 | + |
| 116 | + def collect(self): |
| 117 | + global _app_config_query_failed_once |
| 118 | + row = None |
| 119 | + try: |
| 120 | + row = db.session.execute( |
| 121 | + select(AppConfig).where(AppConfig.key == 'worker_last_iteration_at') |
| 122 | + ).scalar_one_or_none() |
| 123 | + except SQLAlchemyError as e: |
| 124 | + # CRITICAL: rollback the session before continuing. SQLAlchemy 2.x |
| 125 | + # marks the session as failed after a query error; subsequent |
| 126 | + # db.session.execute() calls in the same request (e.g. by the |
| 127 | + # _PendingNotificationsCollector that runs in the same scrape) |
| 128 | + # would otherwise raise PendingRollbackError. |
| 129 | + db.session.rollback() |
| 130 | + # Rate-limit: full traceback only on the first failure per |
| 131 | + # process; subsequent failures get a one-line warning so /metrics |
| 132 | + # scrapes (every 15-30s) on a pre-migration deployment do not |
| 133 | + # flood logs. |
| 134 | + if not _app_config_query_failed_once: |
| 135 | + logger.warning( |
| 136 | + 'Failed to query worker_last_iteration_at from AppConfig; ' |
| 137 | + 'omitting metric', |
| 138 | + exc_info=True, |
| 139 | + ) |
| 140 | + _app_config_query_failed_once = True |
| 141 | + else: |
| 142 | + logger.warning( |
| 143 | + 'Failed to query worker_last_iteration_at from AppConfig: %s: %s', |
| 144 | + type(e).__name__, e, |
| 145 | + ) |
| 146 | + if row is not None: |
| 147 | + try: |
| 148 | + # datetime.fromisoformat accepts only str — None or wrong type |
| 149 | + # raises TypeError, malformed string raises ValueError. Both |
| 150 | + # are non-fatal here; omit the metric and log. |
| 151 | + ts = datetime.fromisoformat(row.value) |
| 152 | + yield GaugeMetricFamily( |
| 153 | + 'esb_worker_last_iteration_timestamp_seconds', |
| 154 | + "Unix timestamp (seconds) of the worker's last successful " |
| 155 | + 'poll. Omitted if the worker has never run or the AppConfig ' |
| 156 | + 'query failed.', |
| 157 | + value=ts.timestamp(), |
| 158 | + ) |
| 159 | + except (ValueError, TypeError): |
| 160 | + logger.warning( |
| 161 | + 'Failed to parse worker last-iteration timestamp value=%r', |
| 162 | + row.value, exc_info=True, |
| 163 | + ) |
| 164 | + |
| 165 | + # Lazy + dotted import inside collect(). Two reasons (NOT circular- |
| 166 | + # import avoidance — there is no circular-import risk): |
| 167 | + # 1. Defers Flask app-context dependencies that some test fixtures |
| 168 | + # lazily install during early test setup. |
| 169 | + # 2. Preserves the monkeypatch surface: tests do |
| 170 | + # monkeypatch.setattr('esb.slack.is_socket_mode_connected', ...) |
| 171 | + # which only takes effect via attribute lookup at call time. A |
| 172 | + # top-of-module `from esb.slack import is_socket_mode_connected` |
| 173 | + # would bind the symbol locally and silently miss the patch. |
| 174 | + # If you "clean this up" by moving the import to module top, you will |
| 175 | + # break the test design. Don't. |
| 176 | + import esb.slack as _slack |
| 177 | + yield GaugeMetricFamily( |
| 178 | + 'esb_socket_mode_enabled', |
| 179 | + '1 if init_slack entered the Socket Mode setup block (tokens set, ' |
| 180 | + 'not TESTING, opt-in flag true). 0 on any of the four early-return ' |
| 181 | + 'paths or if the setup block was never reached.', |
| 182 | + value=1.0 if _slack.is_socket_mode_enabled() else 0.0, |
| 183 | + ) |
| 184 | + yield GaugeMetricFamily( |
| 185 | + 'esb_socket_mode_connected', |
| 186 | + '1 if a Bolt SocketModeHandler is currently bound; 0 if never ' |
| 187 | + 'bound or released. Reflects handler-binding state, NOT live ' |
| 188 | + 'WebSocket connection state — Slack Bolt does not expose mid-life ' |
| 189 | + 'connection callbacks. Transitions 1->0 at process shutdown.', |
| 190 | + value=1.0 if _slack.is_socket_mode_connected() else 0.0, |
| 191 | + ) |
| 192 | + |
| 193 | + |
84 | 194 | def render_metrics() -> tuple[bytes, str]: |
85 | 195 | """Render the Prometheus exposition payload and content-type. |
86 | 196 |
|
87 | 197 | Returns: |
88 | 198 | Tuple of (body, content_type) suitable for a Flask response. |
89 | 199 | """ |
90 | 200 | registry = CollectorRegistry() |
| 201 | + # Order is defensive: with the explicit db.session.rollback() in |
| 202 | + # _WorkerStatusCollector, registration order is no longer load-bearing |
| 203 | + # for correctness, but pinning it makes intent unambiguous. |
91 | 204 | registry.register(_PendingNotificationsCollector()) |
| 205 | + registry.register(_WorkerStatusCollector()) |
92 | 206 | return generate_latest(registry), CONTENT_TYPE_LATEST |
0 commit comments