fitzee commented on code in PR #44009:
URL: https://github.com/apache/superset/pull/44009#discussion_r4001751829
##########
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:
**Hot-path log-spam amplification: `logger.exception` on every commit if
`timing()` ever raises.** This mirrors `incr_capture_error`'s broad-except
pattern, but the frequency is very different: `incr_capture_error` only fires
on error paths, whereas `emit_capture_timing` runs on *every* commit on the
session. If a deployment wires a custom `StatsLogger` without a working
`timing()` (or `stats_logger_manager.instance` is transiently
`None`/misconfigured at startup), every DB commit logs a full traceback — log
flooding at commit throughput. Consider `logger.warning` (message only, no
traceback) or a once/rate-limited log here, since the failure is structural and
identical every time rather than a per-commit anomaly worth a full stack each
time.
##########
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:
**The metric is sold as *the* kill-switch alert signal, but excludes the
dominant capture cost it's meant to gate.** `start = perf_counter()` is set
*after* `session.flush()`, so `finalize.latency` measures only post-flush work.
The `emit_capture_timing` docstring says this series "is the signal an operator
alerts on before flipping [`ENABLE_VERSIONING_CAPTURE`]" — but the kill-switch
removes the *whole* listener, including `capture_initial_states`' per-entity
pre-state SELECTs run in `before_flush`. On a bulk edit of hundreds of
versioned entities, those N SELECTs are plausibly the dominant capture
overhead, and they're entirely outside this timer. So capture can materially
slow saves while `finalize.latency` (even at p99) stays low, and the alert
that's supposed to precede flipping the kill-switch never fires. The finalize
docstring documents the exclusion, but it isn't reconciled with the metric's
stated alerting purpose. Consider also timing the `before_flush` capture stage
(a separate `capture_initial_states.latency` series) so the signal actually
covers the cost the kill-switch removes.
##########
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:
**Fail-open hole: `_current_transaction_id` is unguarded, so an exception
here breaks the user's save.** The module docstring promises "a versioning bug
must never break a user's save," and the inner helpers (`_build_scalar_buffer`,
`_stamp_action_kind_on_transaction`, `_persist_buffered_records`) each swallow
their own exceptions. But `_current_transaction_id` (which does
`session.connection()` + a `sqlalchemy_continuum` lookup) is called directly
inside the `try:`/`finally:` with no `except`, so if it raises — connection
dropped mid-commit, a Continuum internal error — the exception unwinds out of
the `before_commit` listener and fails the commit. This is pre-existing (the
closure had the same shape), but the function is being moved/rewritten here, so
it's a natural moment to wrap the capture body in a swallow (log +
`incr_capture_error`) to actually honor the stated invariant.
##########
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:
**Test gap: timing is never asserted on the real versioned-write path.**
`test_capture_latency_metric_fires_on_commit` adds a `LifecycleRow`
(non-versioned), so `finalize` returns at `tx_id is None` before
`_stamp_action_kind_on_transaction` / `_append_child_records_to_buffer` /
`_persist_buffered_records` ever run. It proves timing fires on a no-op commit,
but the metric's behavior on the actual capture path — the code whose latency
the whole feature exists to measure — is unasserted. A regression that skipped
or mis-timed emission when records are actually persisted would still pass CI.
Worth one test that dirties a versioned entity (buffer non-empty, records
persisted) and asserts a single `finalize.latency` emission covers that path.
--
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]
