eschutho commented on code in PR #42624:
URL: https://github.com/apache/superset/pull/42624#discussion_r3694355996
##########
superset/utils/webdriver.py:
##########
@@ -1051,6 +1452,12 @@ def get_screenshot( # noqa: C901
unexpected_errors,
)
+ if report_execution_context:
+ phase_timeout(
+ "screenshot_capture",
+ None,
+ report_execution_context.post_capture_reserve_seconds,
+ )
img = element.screenshot_as_png
Review Comment:
Real limitation, acknowledged — Selenium's synchronous screenshot command
cannot take a timeout, so it isn't cooperatively bounded (this predates the PR;
Selenium capture has never been bounded). The layered design covers it: the
`phase_timeout` gate stops a capture from *starting* without budget, and if the
capture itself blocks past the deadline, the Celery soft limit (aligned to the
budget) interrupts the task and the command-boundary retry persists the
terminal ERROR inside the 30s hard grace — so the 'terminates before cleanup
completes' consequence doesn't hold. Bounding the blocking WebDriver HTTP call
would need thread/signal wrappers, which we're deliberately not adding here;
Playwright (the primary driver, and the only tiled path) passes an explicit
capture timeout. Note the Bito reply below quotes the Playwright tiled path as
the resolution — different driver; that code is already bounded.
##########
docs/admin_docs/configuration/alerts-reports.mdx:
##########
@@ -244,6 +244,51 @@ class CeleryConfig:
}
CELERY_CONFIG = CeleryConfig
+# Scheduled reports share one deadline across browser readiness, capture/PDF
+# generation, delivery, and terminal-state persistence. The effective budget
+# for a schedule is min(this value, the schedule's working_timeout), so the
+# per-schedule field keeps its meaning as a user-facing cap. The default (one
+# hour) matches the historical working_timeout default, so upgrading changes
+# no default behavior; lower it to enforce a tighter report SLA.
+ALERT_REPORTS_EXECUTION_BUDGET_SECONDS = 3600
+
+# These reserves are part of (not additions to) the total budget and their sum
+# must be less than it. Readiness polling stops in time to leave capacity for
+# the later phases.
+ALERT_REPORTS_EXECUTION_CAPTURE_RESERVE_SECONDS = 60
+ALERT_REPORTS_EXECUTION_DELIVERY_RESERVE_SECONDS = 120
+ALERT_REPORTS_EXECUTION_CLEANUP_RESERVE_SECONDS = 30
+
+# Celery's hard limit leaves this additional window for terminal cleanup after
+# the 15-minute soft limit. ALERT_REPORTS_WORKING_TIME_OUT_KILL controls these
+# Celery limits; disabling it does not disable the application deadline above.
+ALERT_REPORTS_EXECUTION_HARD_TIMEOUT_GRACE_SECONDS = 30
Review Comment:
Correct — stale leftover from the 900s draft. Fixed in `ba7777d94a`: the
comment now describes the grace as following the resolved execution budget
(configured budget capped by working_timeout) instead of naming 15 minutes.
##########
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
+ )
Review Comment:
Agreed the compare-then-commit is not atomic — this is the known TOCTOU
family discussed in the human review above (items 1–2). Without a unique
constraint on `ReportExecutionLog.uuid` or row locking, tightening the compare
only narrows the window, and the schema/locking work (unique index + migration
+ backfill for existing duplicate uuids, guarded writes) is scoped as a tracked
follow-up rather than folded into this PR. Worth noting the check that exists
here is already strictly stronger than pre-PR master, which had no ownership
compare at all on any terminal write path.
##########
superset/commands/report/execute.py:
##########
@@ -639,7 +856,20 @@ def _get_pdf(self) -> bytes:
:raises: ReportSchedulePdfFailedError
"""
screenshots = self._get_screenshots()
+ reserve_seconds = (
+ self._report_execution_context.post_capture_reserve_seconds
+ if self._report_execution_context
+ else 0.0
+ )
+ self._phase_timeout(
+ "pdf_generation",
+ reserve_seconds=reserve_seconds,
+ )
pdf = build_pdf_from_screenshots(screenshots)
+ self._phase_timeout(
+ "pdf_generation",
+ reserve_seconds=reserve_seconds,
+ )
Review Comment:
By design: the deadline is cooperative at operation granularity, and
`build_pdf_from_screenshots` is a synchronous CPU-bound call we intentionally
don't interrupt internally. The backstop is the Celery soft limit, which equals
the budget — if PDF generation blows through the remaining budget,
`SoftTimeLimitExceeded` interrupts the task and the command boundary persists
the terminal ERROR within the 30s hard grace. So 'terminal state persistence
may not execute' doesn't hold: that persistence path is exactly what the
soft-limit envelope exists for. Adding periodic deadline checks inside
PIL-level PDF assembly isn't practical.
##########
superset/commands/report/execute.py:
##########
@@ -1048,10 +1314,17 @@ def _send(
:raises: CommandException
"""
notification_errors: list[SupersetError] = []
+ report_context = getattr(self, "_report_execution_context", None)
for recipient in recipients:
notification = create_notification(recipient, notification_content)
try:
try:
+ log_report_delivery_phase(
+ report_context,
+ getattr(recipient, "type", None),
+ "start",
+ enforce_budget=True,
+ )
Review Comment:
This one misreads the reserve semantics: reserves protect *later* phases, so
the delivery gate reserving only `cleanup_reserve_seconds` is correct — the
delivery reserve is the capacity delivery itself is meant to spend (it's held
back from the earlier readiness/capture phases, see
`readiness_reserve_seconds`/`post_capture_reserve_seconds`). Reserving delivery
capacity *at* the delivery gate would double-count it and starve delivery. An
unbounded `notification.send()` is backstopped the same way as capture: the
per-recipient gate stops sends from starting without budget, and the Celery
soft limit + command-boundary persistence covers a send that blocks.
##########
superset/utils/report_execution.py:
##########
@@ -0,0 +1,269 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Shared deadline and logging context for scheduled report execution."""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+from typing import Any
+from uuid import UUID
+
+logger = logging.getLogger(__name__)
+
+# Minimum working allowance kept above the summed phase reserves when a
+# per-schedule working_timeout would otherwise squeeze the effective budget
+# below what the execution context can represent. A floored budget still
+# fails fast (budget-exceeded on the first phase) rather than erroring while
+# constructing the deadline.
+MIN_REPORT_EXECUTION_WORK_SECONDS = 30.0
+
+
+def validate_report_execution_config(config: Mapping[str, Any]) -> None:
+ """Validate the scheduled-report budget invariant during application
startup."""
+
+ budget = float(config["ALERT_REPORTS_EXECUTION_BUDGET_SECONDS"])
+ reserves = (
+ float(config["ALERT_REPORTS_EXECUTION_CAPTURE_RESERVE_SECONDS"]),
+ float(config["ALERT_REPORTS_EXECUTION_DELIVERY_RESERVE_SECONDS"]),
+ float(config["ALERT_REPORTS_EXECUTION_CLEANUP_RESERVE_SECONDS"]),
+ )
+ hard_timeout_grace = float(
+ config["ALERT_REPORTS_EXECUTION_HARD_TIMEOUT_GRACE_SECONDS"]
+ )
+
+ if budget <= 0:
+ raise ValueError("Report execution budget must be greater than zero")
+ if any(reserve < 0 for reserve in reserves):
+ raise ValueError("Report execution phase reserves cannot be negative")
+ if sum(reserves) >= budget:
+ raise ValueError(
+ "Report execution phase reserves must total less than the
execution budget"
+ )
+ if hard_timeout_grace < 0:
+ raise ValueError("Report execution hard-timeout grace cannot be
negative")
+
+
+def resolve_report_execution_budget_seconds(
+ config: Mapping[str, Any],
+ working_timeout: int | None = None,
+) -> float:
+ """Return the effective execution budget for one REPORT schedule.
+
+ The per-schedule ``working_timeout`` keeps its historical, user-facing
+ meaning ("kill my report after N seconds"): when it is lower than the
+ global ``ALERT_REPORTS_EXECUTION_BUDGET_SECONDS`` it caps the budget, so
+ introducing the global deadline does not silently grant a schedule more
+ time than its owner configured. The result is floored at the summed
+ phase reserves plus a minimal working allowance so the execution context
+ remains constructible; a floored budget fails cleanly on its first phase
+ check instead of raising at setup.
+ """
+ budget = float(config["ALERT_REPORTS_EXECUTION_BUDGET_SECONDS"])
+ if working_timeout is not None:
+ budget = min(budget, float(working_timeout))
+ reserves_total = (
+ float(config["ALERT_REPORTS_EXECUTION_CAPTURE_RESERVE_SECONDS"])
+ + float(config["ALERT_REPORTS_EXECUTION_DELIVERY_RESERVE_SECONDS"])
+ + float(config["ALERT_REPORTS_EXECUTION_CLEANUP_RESERVE_SECONDS"])
+ )
+ if budget < (min_viable := reserves_total +
MIN_REPORT_EXECUTION_WORK_SECONDS):
+ logger.warning(
+ "Report working_timeout=%s is below the minimum viable execution "
+ "budget (%.0fs phase reserves + %.0fs working allowance); "
+ "flooring the effective budget at %.0fs.",
+ working_timeout,
+ reserves_total,
+ MIN_REPORT_EXECUTION_WORK_SECONDS,
+ min_viable,
+ )
+ return min_viable
Review Comment:
Intentional and documented (UPDATING.md and the resolver docstring, with a
runtime warning when the floor engages). The alternative — strictly honoring a
sub-viable working_timeout — means every execution of that schedule fails
deterministically, since the timeout can't fit capture + delivery + cleanup.
Pre-PR, such a schedule was killed mid-capture every run and never delivered
anything, so the floor (reserves + 30s ≈ 240s with defaults) converts a
permanently-broken configuration into a working one while logging the
discrepancy. A schedule owner who genuinely wants sub-4-minute kills has no
deliverable report either way.
--
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]
