mikebridge commented on code in PR #44009:
URL: https://github.com/apache/superset/pull/44009#discussion_r4007259016
##########
superset/versioning/changes/listener.py:
##########
@@ -395,7 +396,51 @@ def _persist_buffered_records(
incr_capture_error("bulk_insert")
-def register_change_record_listener() -> None: # noqa: C901
+def finalize_change_records(session: Session) -> None:
+ """Build and persist the transaction's change records at commit time.
+
+ Module-level (rather than a closure inside the registration function)
+ so the capture write path can be exercised directly by unit tests
+ against an isolated session; it depends only on the session and the
+ module helpers, never on the registered entity classes.
+ """
+ if session.in_nested_transaction() or session.info.get(_FINALIZING_KEY):
+ return
+
+ session.info[_FINALIZING_KEY] = True
+ # Measures the FINALIZE stage only: the timer starts after the flush,
+ # which excludes the transaction's own write cost but also excludes
+ # capture_initial_states' per-entity pre-state SELECTs — and runs
+ # through every capture step and early return. Every commit on the
+ # session emits a sample, including commits touching no versioned
+ # entity, because the whole-listener overhead is exactly what the
+ # kill-switch removes; a flush that raises emits nothing.
+ start: float | None = None
+ try:
+ session.flush()
+ start = perf_counter()
Review Comment:
Agreed, and fixed in 53dcb5b8fd the way you suggested: the before-flush
capture stage now emits its own
`superset.versioning.capture.capture_initial_states.latency` series (sampled
only when at least one versioned entity was captured, so unrelated autoflushes
do not flood it), and `emit_capture_timing`'s docstring now says the two stages
TOGETHER cover the cost the kill-switch removes — alert on both. The loop moved
to a module-level `_capture_initial_states` helper so the stage is unit-tested
(fires with a dirty versioned entity, not otherwise).
##########
superset/versioning/changes/listener.py:
##########
@@ -395,7 +396,51 @@ def _persist_buffered_records(
incr_capture_error("bulk_insert")
-def register_change_record_listener() -> None: # noqa: C901
+def finalize_change_records(session: Session) -> None:
+ """Build and persist the transaction's change records at commit time.
+
+ Module-level (rather than a closure inside the registration function)
+ so the capture write path can be exercised directly by unit tests
+ against an isolated session; it depends only on the session and the
+ module helpers, never on the registered entity classes.
+ """
+ if session.in_nested_transaction() or session.info.get(_FINALIZING_KEY):
+ return
+
+ session.info[_FINALIZING_KEY] = True
+ # Measures the FINALIZE stage only: the timer starts after the flush,
+ # which excludes the transaction's own write cost but also excludes
+ # capture_initial_states' per-entity pre-state SELECTs — and runs
+ # through every capture step and early return. Every commit on the
+ # session emits a sample, including commits touching no versioned
+ # entity, because the whole-listener overhead is exactly what the
+ # kill-switch removes; a flush that raises emits nothing.
+ start: float | None = None
+ try:
+ session.flush()
+ start = perf_counter()
+ initial_states: dict[tuple[str, int], tuple[Any, dict[str, Any]]] = (
+ session.info.get(_INITIAL_STATES_KEY, {})
+ )
+ buffer = _build_scalar_buffer(initial_states)
+
+ tx_id = _current_transaction_id(session)
Review Comment:
Fixed in 53dcb5b8fd — applied the guard verbatim (log, count as
`transaction_lookup`, return). Added a test that makes
`_current_transaction_id` raise and asserts the commit still succeeds and the
user's row persists, with the error counted — the probe result, now pinned.
Thanks both.
##########
superset/versioning/metrics.py:
##########
@@ -52,3 +52,28 @@ def incr_capture_error(stage: str) -> None:
stats_logger_manager.instance.incr(f"{_CAPTURE_METRIC_PREFIX}.{stage}.error")
except Exception: # pylint: disable=broad-except
logger.exception("versioning: failed to emit capture-error metric")
+
+
+def emit_capture_timing(stage: str, duration_ms: float) -> None:
+ """Emit the write-path latency for one capture *stage*, in milliseconds.
+
+ The documented recovery lever for capture trouble is the
+ ``ENABLE_VERSIONING_CAPTURE`` kill-switch, flipped on save-path
+ slowdown — this series
+ (``superset.versioning.capture.<stage>.latency``) is the signal an
+ operator alerts on before flipping it. :func:`incr_capture_error`
+ covers *loss*; this covers *slowdown*. The series includes every
+ commit on the session — commits touching no versioned entity still
+ pay the listener overhead — so alerts belong on upper percentiles,
+ not the mean. Best-effort under the same fail-open posture: metrics
+ emission must never itself break a user's save.
+ """
+ # pylint: disable=import-outside-toplevel
+ try:
+ from superset.extensions import stats_logger_manager
+
+ stats_logger_manager.instance.timing(
+ f"{_CAPTURE_METRIC_PREFIX}.{stage}.latency", duration_ms
+ )
+ except Exception: # pylint: disable=broad-except
+ logger.exception("versioning: failed to emit capture-latency metric")
Review Comment:
Fixed in 53dcb5b8fd — downgraded to one `logger.warning` line (message only,
no traceback) per occurrence, with the hot-path rationale in a comment; the
fail-open test now asserts `warning` is called and `exception` is not.
##########
tests/unit_tests/versioning/test_listener.py:
##########
@@ -250,3 +251,94 @@ def test_transient_persist_failure_is_logged_and_counted(
log_spy.assert_called_once()
metric_spy.assert_called_once_with("bulk_insert")
+
+
+def test_capture_latency_metric_fires_on_commit(
+ lifecycle_session: Session, mocker: Any
+) -> None:
+ """The finalizer emits the write-path latency series on every save-path
+ commit — the kill-switch's own decision signal, measuring capture
+ overhead only (the timer starts after the transaction's own flush).
+ Driven through the real module-level finalizer on an isolated session
+ (no versioning tables needed: the tx-id early return still passes the
+ timing's ``finally``)."""
+ sa.event.listen(
+ lifecycle_session, "before_commit", listener.finalize_change_records
+ )
+ manager = MagicMock()
+ mocker.patch("superset.extensions.stats_logger_manager", manager)
+ lifecycle_session.add(LifecycleRow(value="timed"))
Review Comment:
Added in 53dcb5b8fd —
`test_capture_latency_metric_fires_once_on_the_versioned_write_path` retains a
versioned entity's initial state (non-empty buffer), resolves a tx id, and
asserts the records reach persistence AND exactly one `finalize.latency` sample
is emitted for that real path, so a regression that skipped or mis-timed
emission when records are actually persisted would fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
