eschutho commented on code in PR #42624:
URL: https://github.com/apache/superset/pull/42624#discussion_r3694422960
##########
superset/utils/webdriver.py:
##########
@@ -380,11 +440,15 @@ def _wait_for_charts_ready(
elapsed,
context_suffix,
)
- readiness_predicate = (
- CHART_CONTAINER_READY_JS
- if element_name == "chart-container"
- else CHART_HOLDERS_READY_JS
- )
+ if element_name == "chart-container":
+ readiness_predicate = CHART_CONTAINER_READY_JS
+ elif report_execution_context:
+ readiness_predicate = REPORT_CHART_HOLDERS_READY_JS
Review Comment:
Real trade-off, deliberately carried over rather than introduced here — the
viewport-scoped predicate came from #42153/#42253's production-safe readiness
work, because requiring below-the-fold holders deadlocks on virtualized
dashboards (off-screen holders never render by design). Full-dashboard
readiness is exactly what the tiled path provides: it scrolls tile by tile and
validates each viewport before capture, and the thresholds
(`SCREENSHOT_TILED_CHART_THRESHOLD`, `SCREENSHOT_TILED_HEIGHT_THRESHOLD`) route
chart-heavy/tall dashboards there. The residual window is a dashboard taller
than the browser window but under both tiling thresholds; operators can close
it by lowering `SCREENSHOT_TILED_HEIGHT_THRESHOLD` toward the viewport height.
Auto-tiling whenever element height exceeds the viewport is a reasonable future
tightening, but it changes capture behavior for a class of currently-working
dashboards, so it's out of scope for this PR.
##########
superset/commands/report/execute.py:
##########
@@ -123,6 +131,137 @@ def resolve_executor_user(model: ReportSchedule) ->
tuple["User", str]:
return user, username
+def log_report_delivery_phase(
+ report_context: ReportExecutionContext | None,
+ recipient_type: ReportRecipientType | None,
+ phase: str,
+ *,
+ enforce_budget: bool,
+) -> None:
+ """Enforce and log a notification phase when executing a report."""
+
+ if report_context is None:
+ return
+ deadline = report_context.deadline
+ if enforce_budget:
+ deadline.timeout_seconds(
+ "notification_delivery",
+ reserve_seconds=report_context.cleanup_reserve_seconds,
+ )
+ logger.info(
+ "report_delivery_%s %s recipient_type=%s elapsed_seconds=%.2f "
+ "remaining_seconds=%.2f",
+ phase,
+ report_context.log_context,
+ recipient_type,
+ deadline.elapsed_seconds,
+ deadline.remaining_seconds,
+ )
+
+
+def persist_owned_report_execution_terminal_error(
+ report_schedule_id: int,
+ execution_id: UUID,
+ error_message: str,
+ terminal_reason: str,
+ report_context: ReportExecutionContext | None = None,
+) -> bool:
+ """
+ Terminalize this command's WORKING row from its application-owned boundary.
+
+ Report states normally persist their terminal result before re-raising. If
+ that first write loses its transaction or database connection, the command
+ boundary is the last safe in-process retry: it still has Flask application
+ context and knows the execution UUID it owns. A compare against the latest
+ active WORKING row prevents an old worker from changing the schedule state
+ after a newer execution has started.
+ """
+
+ try:
+ # The state-machine transaction has already rolled back on its way to
+ # this boundary. Roll back again so a failed terminal flush cannot
leave
+ # the scoped session unusable for the retry.
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ working_log = (
+ db.session.query(ReportExecutionLog)
+ .filter(
+ ReportExecutionLog.report_schedule_id == report_schedule_id,
+ ReportExecutionLog.uuid == execution_id,
+ ReportExecutionLog.state == ReportState.WORKING,
+ ReportExecutionLog.error_message.is_(None),
+ )
+ .first()
+ )
+ if working_log is None:
+ return False
+
+ latest_working_log = (
+ db.session.query(ReportExecutionLog)
+ .filter(
+ ReportExecutionLog.report_schedule_id == report_schedule_id,
+ ReportExecutionLog.state == ReportState.WORKING,
+ ReportExecutionLog.error_message.is_(None),
+ )
+ .order_by(ReportExecutionLog.end_dttm.desc())
+ .first()
+ )
+ report_schedule = working_log.report_schedule
+ owns_schedule_state = (
+ report_schedule.last_state == ReportState.WORKING
+ and latest_working_log is not None
+ and latest_working_log.uuid == execution_id
+ )
+ ended_at = datetime.now(timezone.utc).replace(tzinfo=None)
+ working_log.state = ReportState.ERROR
+ working_log.error_message = error_message
+ working_log.end_dttm = ended_at
+ if owns_schedule_state:
+ report_schedule.last_state = ReportState.ERROR
+ report_schedule.last_eval_dttm = ended_at
+
+ db.session.commit() # pylint: disable=consider-using-transaction
Review Comment:
Same finding as the thread on line 213 (and items 1–2 of the human review
above): agreed the compare-then-commit isn't atomic, and a real fix needs a
unique constraint on `ReportExecutionLog.uuid` plus a guarded/locked write —
schema work that's scoped as a tracked follow-up. The compare here is strictly
stronger than pre-PR master, which wrote terminal state with no ownership check
at all.
##########
superset/commands/report/execute.py:
##########
@@ -1472,11 +1808,62 @@ def __init__(self, task_id: str, model_id: int,
scheduled_dttm: datetime):
self._execution_id = UUID(task_id)
def run(self) -> None:
+ monotonic_started_at = time.monotonic()
+ report_execution_context: ReportExecutionContext | None = None
+ owns_report_working_state = False
try:
self.validate()
if not self._model:
raise ReportScheduleExecuteUnexpectedError()
+ if self._model.type == ReportScheduleType.REPORT:
+ # An invocation that enters on WORKING is a duplicate or stale
+ # recovery, not the owner that created the active row. Its
state
+ # handler may terminalize a stale execution, but the command
+ # boundary must never infer ownership from a replayed UUID.
+ owns_report_working_state = (
+ self._model.last_state != ReportState.WORKING
+ )
+ total_seconds = resolve_report_execution_budget_seconds(
+ app.config,
+ working_timeout=self._model.working_timeout,
+ )
Review Comment:
Same as the resolver thread in report_execution.py (a review instruction was
saved there): the floor is intentional and documented in UPDATING.md, with a
runtime warning when it engages. Strictly honoring a sub-viable working_timeout
(one below capture+delivery+cleanup reserves) means that schedule fails
deterministically on every run — pre-PR it was killed mid-capture and never
delivered anything. The floor converts a permanently-broken configuration into
a working one; rejecting such configs at save time is a fair alternative but is
an API/validation change beyond this PR.
##########
superset/tasks/scheduler.py:
##########
@@ -98,19 +100,14 @@ def scheduler(self: Task) -> None: # pylint:
disable=unused-argument
triggered_at, active_schedule.crontab, active_schedule.timezone
):
logger.info("Scheduling alert %s eta: %s", active_schedule.name,
schedule)
- async_options = {"eta": schedule}
- if (
- active_schedule.working_timeout is not None
- and current_app.config["ALERT_REPORTS_WORKING_TIME_OUT_KILL"]
- ):
- async_options["time_limit"] = (
- active_schedule.working_timeout
- + current_app.config["ALERT_REPORTS_WORKING_TIME_OUT_LAG"]
- )
- async_options["soft_time_limit"] = (
- active_schedule.working_timeout
- +
current_app.config["ALERT_REPORTS_WORKING_SOFT_TIME_OUT_LAG"]
- )
+ async_options = {
+ "eta": schedule,
+ **get_report_task_timeout_options(
+ is_report=active_schedule.type ==
ReportScheduleType.REPORT,
+ working_timeout=active_schedule.working_timeout,
+ config=current_app.config,
+ ),
+ }
execute.apply_async((active_schedule.id,), **async_options)
Review Comment:
Duplicate of the resolver-floor finding (see the report_execution.py thread)
— same intentional behavior, same UPDATING.md documentation, evaluated once in
`resolve_report_execution_budget_seconds` and consumed consistently by both the
Celery limit derivation here and the in-process deadline, so the two never
disagree.
##########
superset/utils/screenshot_utils.py:
##########
@@ -406,16 +515,12 @@ def _raise_if_budget_exhausted(elapsed: float,
remaining_budget: float) -> None:
# Wait for scroll to settle and content to load
page.wait_for_timeout(SCROLL_SETTLE_TIMEOUT_MS)
- # Recompute the remaining budget after the scroll-settle sleep --
- # which itself consumes real wall-clock time -- rather than
- # reusing the value from before it, so the readiness-check
- # timeout below is capped against a fresh number instead of a
- # stale one that would let each tile overrun the budget by up
- # to one settle interval.
- tile_wait_start = time.monotonic()
- elapsed = tile_wait_start - start_time
- remaining_budget = wait_budget_seconds - elapsed
- _raise_if_budget_exhausted(elapsed, remaining_budget)
+ # Re-check after the scroll-settle sleep -- which itself consumes
+ # real wall-clock time -- so the readiness-check timeout below is
+ # derived from a fresh remaining value instead of a stale one
+ # that would let each tile overrun the budget by up to one settle
+ # interval (_timeout_seconds also recomputes at call time).
+ _raise_if_budget_exhausted()
Review Comment:
The premise isn't quite right: there's a budget check immediately *before*
the scroll (`_raise_if_budget_exhausted()` above the `scrollTo`), and a second
check immediately *after* the settle sleep — added precisely so the per-tile
readiness timeout derives from a fresh remaining value (see the comment above
that second check). The worst case is one settle interval (1s) of overshoot
past the deadline before the re-check raises, which is absorbed by the 30s
cleanup reserve plus the 30s Celery hard grace. Gating the sleep itself on
remaining time would save at most that 1s in an execution that is already
failing.
##########
superset/commands/report/execute.py:
##########
@@ -123,6 +131,137 @@ def resolve_executor_user(model: ReportSchedule) ->
tuple["User", str]:
return user, username
+def log_report_delivery_phase(
+ report_context: ReportExecutionContext | None,
+ recipient_type: ReportRecipientType | None,
+ phase: str,
+ *,
+ enforce_budget: bool,
+) -> None:
+ """Enforce and log a notification phase when executing a report."""
+
+ if report_context is None:
+ return
+ deadline = report_context.deadline
+ if enforce_budget:
+ deadline.timeout_seconds(
+ "notification_delivery",
+ reserve_seconds=report_context.cleanup_reserve_seconds,
+ )
+ logger.info(
+ "report_delivery_%s %s recipient_type=%s elapsed_seconds=%.2f "
+ "remaining_seconds=%.2f",
+ phase,
+ report_context.log_context,
+ recipient_type,
+ deadline.elapsed_seconds,
+ deadline.remaining_seconds,
+ )
+
+
Review Comment:
Added in `7e2010fe0d`: three unit tests covering the no-context no-op, the
enforce_budget=True raise on an exhausted budget, and the enforce_budget=False
post-send logging path (which must record the phase rather than raise
mid-notification).
##########
superset/commands/report/execute.py:
##########
@@ -123,6 +131,137 @@ def resolve_executor_user(model: ReportSchedule) ->
tuple["User", str]:
return user, username
+def log_report_delivery_phase(
+ report_context: ReportExecutionContext | None,
+ recipient_type: ReportRecipientType | None,
+ phase: str,
+ *,
+ enforce_budget: bool,
+) -> None:
+ """Enforce and log a notification phase when executing a report."""
+
+ if report_context is None:
+ return
+ deadline = report_context.deadline
+ if enforce_budget:
+ deadline.timeout_seconds(
+ "notification_delivery",
+ reserve_seconds=report_context.cleanup_reserve_seconds,
+ )
+ logger.info(
+ "report_delivery_%s %s recipient_type=%s elapsed_seconds=%.2f "
+ "remaining_seconds=%.2f",
+ phase,
+ report_context.log_context,
+ recipient_type,
+ deadline.elapsed_seconds,
+ deadline.remaining_seconds,
+ )
+
+
+def persist_owned_report_execution_terminal_error(
+ report_schedule_id: int,
+ execution_id: UUID,
+ error_message: str,
+ terminal_reason: str,
+ report_context: ReportExecutionContext | None = None,
+) -> bool:
+ """
+ Terminalize this command's WORKING row from its application-owned boundary.
+
+ Report states normally persist their terminal result before re-raising. If
+ that first write loses its transaction or database connection, the command
+ boundary is the last safe in-process retry: it still has Flask application
+ context and knows the execution UUID it owns. A compare against the latest
+ active WORKING row prevents an old worker from changing the schedule state
+ after a newer execution has started.
+ """
+
+ try:
+ # The state-machine transaction has already rolled back on its way to
+ # this boundary. Roll back again so a failed terminal flush cannot
leave
+ # the scoped session unusable for the retry.
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ working_log = (
+ db.session.query(ReportExecutionLog)
+ .filter(
+ ReportExecutionLog.report_schedule_id == report_schedule_id,
+ ReportExecutionLog.uuid == execution_id,
+ ReportExecutionLog.state == ReportState.WORKING,
+ ReportExecutionLog.error_message.is_(None),
+ )
+ .first()
+ )
+ if working_log is None:
+ return False
+
+ latest_working_log = (
+ db.session.query(ReportExecutionLog)
+ .filter(
+ ReportExecutionLog.report_schedule_id == report_schedule_id,
+ ReportExecutionLog.state == ReportState.WORKING,
+ ReportExecutionLog.error_message.is_(None),
+ )
+ .order_by(ReportExecutionLog.end_dttm.desc())
+ .first()
+ )
+ report_schedule = working_log.report_schedule
+ owns_schedule_state = (
+ report_schedule.last_state == ReportState.WORKING
+ and latest_working_log is not None
+ and latest_working_log.uuid == execution_id
+ )
+ ended_at = datetime.now(timezone.utc).replace(tzinfo=None)
+ working_log.state = ReportState.ERROR
+ working_log.error_message = error_message
+ working_log.end_dttm = ended_at
+ if owns_schedule_state:
+ report_schedule.last_state = ReportState.ERROR
+ report_schedule.last_eval_dttm = ended_at
+
+ db.session.commit() # pylint: disable=consider-using-transaction
+ log_context = (
+ report_context.log_context
+ if report_context is not None
+ else (
+ f"capture_kind=report execution_id={execution_id} "
+ f"report_schedule_id={report_schedule_id} "
+ f"dashboard_id={report_schedule.dashboard_id} "
+ f"chart_id={report_schedule.chart_id}"
+ )
+ )
+ elapsed_seconds = (
+ f"{report_context.deadline.elapsed_seconds:.2f}"
+ if report_context is not None
+ else "unknown"
+ )
+ remaining_seconds = (
+ f"{report_context.deadline.remaining_seconds:.2f}"
+ if report_context is not None
+ else "unknown"
+ )
+ logger.info(
+ "report_execution_terminal %s state=%s terminal_reason=%s "
+ "elapsed_seconds=%s remaining_seconds=%s",
+ log_context,
+ ReportState.ERROR.value,
+ terminal_reason,
+ elapsed_seconds,
+ remaining_seconds,
+ )
+ return True
+ except Exception: # noqa: BLE001 # never mask the report's original
exception
+ db.session.rollback() # pylint: disable=consider-using-transaction
+ logger.exception(
+ "Failed terminal persistence retry for report execution "
+ "capture_kind=report execution_id=%s report_schedule_id=%s "
+ "terminal_reason=terminal_persistence_retry_failed",
+ execution_id,
+ report_schedule_id,
+ )
+ return False
Review Comment:
Added in `7e2010fe0d`:
`test_terminal_persistence_retry_survives_database_failure` makes the session
query raise, and asserts the double rollback (entry + handler), the
`terminal_persistence_retry_failed` exception log, no commit, and the `False`
return that keeps the report's original exception unmasked.
##########
superset/utils/webdriver.py:
##########
@@ -982,57 +1247,193 @@ def get_screenshot( # noqa: C901
logger.debug(
"Wait for the presence of %s at url: %s", element_name, url
)
- element = WebDriverWait(driver,
self._screenshot_locate_wait).until(
- EC.presence_of_element_located((By.CLASS_NAME,
element_name))
- )
+ element = WebDriverWait(
+ driver,
+ phase_timeout(
+ "dashboard_mount",
+ self._screenshot_locate_wait,
+ (
+ report_execution_context.readiness_reserve_seconds
+ if report_execution_context
+ else 0.0
+ ),
+ ),
+ ).until(EC.presence_of_element_located((By.CLASS_NAME,
element_name)))
except TimeoutException:
logger.warning(
"Selenium timed out requesting url %s", url, exc_info=True
)
raise
- try:
- # chart containers didn't render
- logger.debug("Wait for chart containers to draw at url: %s",
url)
- WebDriverWait(driver, self._screenshot_locate_wait).until(
- EC.visibility_of_all_elements_located(
- (By.CLASS_NAME, "chart-container")
- )
+ if report_execution_context and element_name in {
+ "standalone",
+ "chart-container",
+ }:
+ readiness_predicate = (
+ REPORT_CHART_HOLDERS_READY_JS
+ if element_name == "standalone"
+ else CHART_CONTAINER_READY_JS
+ )
+ readiness_timeout = phase_timeout(
+ "chart_readiness",
+ None,
+ report_execution_context.readiness_reserve_seconds,
)
- except TimeoutException:
- logger.info("Timeout Exception caught")
- # Fallback to allow a screenshot of an empty dashboard
try:
- WebDriverWait(driver, 0).until(
+ WebDriverWait(driver, readiness_timeout).until(
+ lambda webdriver: webdriver.execute_script(
+ f"return ({readiness_predicate})()"
+ )
+ )
+ holder_states = (
+ driver.execute_script(
+ f"return ({FIND_CHART_HOLDER_STATES_JS})()"
+ )
+ if element_name == "standalone"
+ else [
+ {
+ "chartId": report_execution_context.chart_id,
+ "state": "rendered",
+ }
+ ]
+ )
+ ready_states = {"rendered", "empty", "error",
"virtualized"}
+ deadline = report_execution_context.deadline
+ logger.info(
+ "report_readiness_ready url=%s expected_holders=%s "
+ "mounted_holders=%s ready_holders=%s
elapsed_seconds=%s "
+ "remaining_seconds=%s%s",
+ url,
+ report_execution_context.expected_chart_count,
+ len(holder_states),
+ sum(
+ holder.get("state") in ready_states
+ for holder in holder_states
+ ),
+ (f"{deadline.elapsed_seconds:.2f}" if deadline else
None),
Review Comment:
Fixed in `7e2010fe0d` — both this and the second occurrence at line 1345
removed; `deadline` is assigned unconditionally there and
`ReportExecutionDeadline` is always truthy.
##########
superset/utils/webdriver.py:
##########
@@ -982,57 +1247,193 @@ def get_screenshot( # noqa: C901
logger.debug(
"Wait for the presence of %s at url: %s", element_name, url
)
- element = WebDriverWait(driver,
self._screenshot_locate_wait).until(
- EC.presence_of_element_located((By.CLASS_NAME,
element_name))
- )
+ element = WebDriverWait(
+ driver,
+ phase_timeout(
+ "dashboard_mount",
+ self._screenshot_locate_wait,
+ (
+ report_execution_context.readiness_reserve_seconds
+ if report_execution_context
+ else 0.0
+ ),
+ ),
+ ).until(EC.presence_of_element_located((By.CLASS_NAME,
element_name)))
except TimeoutException:
logger.warning(
"Selenium timed out requesting url %s", url, exc_info=True
)
raise
- try:
- # chart containers didn't render
- logger.debug("Wait for chart containers to draw at url: %s",
url)
- WebDriverWait(driver, self._screenshot_locate_wait).until(
- EC.visibility_of_all_elements_located(
- (By.CLASS_NAME, "chart-container")
- )
+ if report_execution_context and element_name in {
+ "standalone",
+ "chart-container",
+ }:
+ readiness_predicate = (
+ REPORT_CHART_HOLDERS_READY_JS
+ if element_name == "standalone"
+ else CHART_CONTAINER_READY_JS
+ )
+ readiness_timeout = phase_timeout(
+ "chart_readiness",
+ None,
+ report_execution_context.readiness_reserve_seconds,
)
- except TimeoutException:
- logger.info("Timeout Exception caught")
- # Fallback to allow a screenshot of an empty dashboard
try:
- WebDriverWait(driver, 0).until(
+ WebDriverWait(driver, readiness_timeout).until(
+ lambda webdriver: webdriver.execute_script(
+ f"return ({readiness_predicate})()"
+ )
+ )
+ holder_states = (
+ driver.execute_script(
+ f"return ({FIND_CHART_HOLDER_STATES_JS})()"
+ )
+ if element_name == "standalone"
+ else [
+ {
+ "chartId": report_execution_context.chart_id,
+ "state": "rendered",
+ }
+ ]
+ )
+ ready_states = {"rendered", "empty", "error",
"virtualized"}
+ deadline = report_execution_context.deadline
+ logger.info(
+ "report_readiness_ready url=%s expected_holders=%s "
+ "mounted_holders=%s ready_holders=%s
elapsed_seconds=%s "
+ "remaining_seconds=%s%s",
+ url,
+ report_execution_context.expected_chart_count,
+ len(holder_states),
+ sum(
+ holder.get("state") in ready_states
+ for holder in holder_states
+ ),
+ (f"{deadline.elapsed_seconds:.2f}" if deadline else
None),
+ (f"{deadline.remaining_seconds:.2f}" if deadline else
None),
+ f" [{log_context}]" if log_context else "",
+ )
+ except TimeoutException:
+ holder_states = (
+ driver.execute_script(
+ f"return ({FIND_CHART_HOLDER_STATES_JS})()"
+ )
+ if element_name == "standalone"
+ else [
+ {
+ "chartId": report_execution_context.chart_id,
+ "state": "not_ready",
+ }
+ ]
+ )
+ ready_states = {"rendered", "empty", "error",
"virtualized"}
+ ready_holders = sum(
+ holder.get("state") in ready_states for holder in
holder_states
+ )
+ deadline = report_execution_context.deadline
+ logger.warning(
+ "report_readiness_terminal url=%s expected_holders=%s "
+ "mounted_holders=%s ready_holders=%s
elapsed_seconds=%s "
+ "remaining_seconds=%s effective_wait_seconds=%.2f%s "
+ "terminal_reason=readiness_timeout states=%s; "
+ "aborting before capture or delivery",
+ url,
+ report_execution_context.expected_chart_count,
+ len(holder_states),
+ ready_holders,
+ (f"{deadline.elapsed_seconds:.2f}" if deadline else
None),
Review Comment:
Fixed in `7e2010fe0d` (same commit as the sibling at line 1313).
--
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]
