msyavuz commented on code in PR #44144:
URL: https://github.com/apache/superset/pull/44144#discussion_r4046185196
##########
superset/dashboards/api.py:
##########
@@ -1965,44 +1984,249 @@ def cache_dashboard_screenshot(self, pk: int,
**kwargs: Any) -> WerkzeugResponse
dashboard_url = get_url_path("Superset.dashboard_permalink",
key=permalink_key)
screenshot_obj = DashboardScreenshot(dashboard_url, dashboard.digest)
- cache_key = screenshot_obj.get_cache_key(window_size, thumb_size,
permalink_key)
- image_url = get_url_path(
- "DashboardRestApi.screenshot", pk=dashboard.id, digest=cache_key
- )
- cache_payload = (
- screenshot_obj.get_from_cache_key(cache_key) or
ScreenshotCachePayload()
+ cache_scope = f"dashboard:{dashboard.id}"
+ request_cache_key = screenshot_obj.get_api_request_cache_key(
+ window_size,
+ thumb_size,
+ permalink_key,
+ cache_scope,
)
- def build_response(status_code: int) -> WerkzeugResponse:
+ def build_response(
+ status_code: int,
+ cache_key: str,
+ cache_payload: ScreenshotCachePayload,
+ ) -> WerkzeugResponse:
return self.response(
status_code,
cache_key=cache_key,
dashboard_url=dashboard_url,
- image_url=image_url,
+ image_url=get_url_path(
+ "DashboardRestApi.screenshot",
+ pk=dashboard.id,
+ digest=cache_key,
+ ),
+ task_timeout_seconds=(
+ 2 * current_app.config["THUMBNAIL_COMPUTING_CACHE_TTL"]
+ ),
task_updated_at=cache_payload.get_timestamp(),
task_status=cache_payload.get_status(),
)
- if cache_payload.should_trigger_task(
- force, expected_scope=f"dashboard:{dashboard.id}"
- ):
- logger.info("Triggering screenshot ASYNC")
- cache_dashboard_screenshot.delay(
- username=get_current_user(),
- guest_token=(
- g.user.guest_token
- if get_current_user() and isinstance(g.user, GuestUser)
+ def get_current_generation() -> tuple[
+ str | None, ScreenshotCachePayload | None
+ ]:
+ cache_key = screenshot_obj.get_current_api_generation_cache_key(
+ request_cache_key,
+ cache_scope,
+ )
+ return (
+ cache_key,
+ (
+ screenshot_obj.get_from_cache_key(
+ cache_key,
+ raise_on_error=True,
+ )
+ if cache_key
else None
),
- dashboard_id=dashboard.id,
- dashboard_url=dashboard_url,
- thumb_size=thumb_size,
- window_size=window_size,
- cache_key=cache_key,
- force=force,
)
- return build_response(202)
- return build_response(200)
+
+ try:
+ observed_cache_key, _ = get_current_generation()
+ except ScreenshotCacheError:
+ logger.exception("Screenshot cache read failed: %s",
request_cache_key)
+ return self.response(
+ 503,
+ message=gettext("Screenshot cache is unavailable"),
+ )
+
+ producer_lock_ttl = get_default_lock_ttl()
+ # A contender must be able to outwait a live producer's lease. In
+ # particular, Celery broker publication can legitimately exceed one
+ # second; timing out sooner would reject a caller just before the
+ # producer publishes the generation it should join.
+ lock_deadline = (
Review Comment:
If a producer dies between acquiring the lock and publishing, every later
POST for that dashboard now holds a sync worker for the full
`DISTRIBUTED_LOCK_DEFAULT_TTL` (30s) before 503-ing, where master returned
immediately — enough concurrent download clicks could starve a sync/gthread
pool.
##########
superset/utils/screenshots.py:
##########
@@ -191,13 +201,46 @@ def is_error_cache_ttl_expired(self) -> bool:
def is_computing_stale(self) -> bool:
"""Check if a COMPUTING status is stale (task likely failed or
stuck)."""
+ return self.is_in_progress_stale()
+
+ def is_in_progress_stale(self) -> bool:
+ """Check if a pending or computing request has exceeded its lease."""
computing_ttl = app.config["THUMBNAIL_COMPUTING_CACHE_TTL"]
return (
datetime.now() - datetime.fromisoformat(self.get_timestamp())
).total_seconds() >= computing_ttl
- def should_trigger_task(
+ def is_in_progress(self) -> bool:
+ """Return whether screenshot computation has not reached a terminal
state."""
+
+ return self.status in (StatusValues.PENDING, StatusValues.COMPUTING)
+
+ def is_updated(self) -> bool:
+ """Return whether screenshot computation completed successfully."""
+
+ return self.status == StatusValues.UPDATED
+
+ def should_enqueue_task(
self, force: bool = False, expected_scope: str | None = None
+ ) -> bool:
+ """Return whether an API producer should enqueue a new generation.
+
+ Fresh pending/computing state is already accepted work, so even forced
+ callers observe it instead of producing another generation. A stale
+ in-progress state remains retryable through the existing lease TTL.
+ """
+
+ if expected_scope is not None and self._scope != expected_scope:
+ return True
+ if self.is_in_progress():
+ return self.is_in_progress_stale()
Review Comment:
`force` is ignored while an in-progress generation is fresh, so if the
worker dies or the broker drops the message after `Pending` is persisted, a
forced retry rejoins the dead generation for up to
`THUMBNAIL_COMPUTING_CACHE_TTL` (360s) instead of re-enqueueing as it did
before. Intended?
##########
superset/dashboards/api.py:
##########
@@ -1965,44 +1984,249 @@ def cache_dashboard_screenshot(self, pk: int,
**kwargs: Any) -> WerkzeugResponse
dashboard_url = get_url_path("Superset.dashboard_permalink",
key=permalink_key)
screenshot_obj = DashboardScreenshot(dashboard_url, dashboard.digest)
- cache_key = screenshot_obj.get_cache_key(window_size, thumb_size,
permalink_key)
- image_url = get_url_path(
- "DashboardRestApi.screenshot", pk=dashboard.id, digest=cache_key
- )
- cache_payload = (
- screenshot_obj.get_from_cache_key(cache_key) or
ScreenshotCachePayload()
+ cache_scope = f"dashboard:{dashboard.id}"
+ request_cache_key = screenshot_obj.get_api_request_cache_key(
+ window_size,
+ thumb_size,
+ permalink_key,
+ cache_scope,
)
- def build_response(status_code: int) -> WerkzeugResponse:
+ def build_response(
+ status_code: int,
+ cache_key: str,
+ cache_payload: ScreenshotCachePayload,
+ ) -> WerkzeugResponse:
return self.response(
status_code,
cache_key=cache_key,
dashboard_url=dashboard_url,
- image_url=image_url,
+ image_url=get_url_path(
+ "DashboardRestApi.screenshot",
+ pk=dashboard.id,
+ digest=cache_key,
+ ),
+ task_timeout_seconds=(
+ 2 * current_app.config["THUMBNAIL_COMPUTING_CACHE_TTL"]
+ ),
task_updated_at=cache_payload.get_timestamp(),
task_status=cache_payload.get_status(),
)
- if cache_payload.should_trigger_task(
- force, expected_scope=f"dashboard:{dashboard.id}"
- ):
- logger.info("Triggering screenshot ASYNC")
- cache_dashboard_screenshot.delay(
- username=get_current_user(),
- guest_token=(
- g.user.guest_token
- if get_current_user() and isinstance(g.user, GuestUser)
+ def get_current_generation() -> tuple[
+ str | None, ScreenshotCachePayload | None
+ ]:
+ cache_key = screenshot_obj.get_current_api_generation_cache_key(
+ request_cache_key,
+ cache_scope,
+ )
+ return (
+ cache_key,
+ (
+ screenshot_obj.get_from_cache_key(
+ cache_key,
+ raise_on_error=True,
+ )
+ if cache_key
else None
),
- dashboard_id=dashboard.id,
- dashboard_url=dashboard_url,
- thumb_size=thumb_size,
- window_size=window_size,
- cache_key=cache_key,
- force=force,
)
- return build_response(202)
- return build_response(200)
+
+ try:
+ observed_cache_key, _ = get_current_generation()
+ except ScreenshotCacheError:
+ logger.exception("Screenshot cache read failed: %s",
request_cache_key)
+ return self.response(
+ 503,
+ message=gettext("Screenshot cache is unavailable"),
+ )
+
+ producer_lock_ttl = get_default_lock_ttl()
+ # A contender must be able to outwait a live producer's lease. In
+ # particular, Celery broker publication can legitimately exceed one
+ # second; timing out sooner would reject a caller just before the
+ # producer publishes the generation it should join.
+ lock_deadline = (
+ time.monotonic() + producer_lock_ttl +
SCREENSHOT_API_LOCK_RETRY_SECONDS
+ )
+ while True:
+ lock_response: WerkzeugResponse | None = None
+ try:
+ with DistributedLock(
+ namespace=SCREENSHOT_API_LOCK_NAMESPACE,
+ request_cache_key=request_cache_key,
+ ttl_seconds=producer_lock_ttl,
+ ):
+ try:
+ cache_key, cached_payload = get_current_generation()
+ except ScreenshotCacheError:
+ logger.exception(
+ "Screenshot cache read failed: %s",
+ request_cache_key,
+ )
+ lock_response = self.response(
+ 503,
+ message=gettext("Screenshot cache is unavailable"),
+ )
+ return lock_response
+
+ cache_payload = cached_payload or ScreenshotCachePayload(
+ scope=cache_scope
+ )
+ if cached_payload is not None:
+ if cache_key is None:
+ logger.error(
+ "Screenshot generation payload has no cache
key: %s",
+ request_cache_key,
+ )
+ lock_response = self.response(
+ 503,
+ message=gettext("Screenshot cache is
unavailable"),
+ )
+ return lock_response
+ if (
+ cache_key != observed_cache_key
+ or not cache_payload.should_enqueue_task(
+ force,
+ expected_scope=cache_scope,
+ )
+ ):
+ lock_response = build_response(
+ 200, cache_key, cache_payload
+ )
+ return lock_response
+
+ logger.info("Triggering screenshot ASYNC")
+ next_cache_key =
screenshot_obj.get_next_api_generation_cache_key(
+ request_cache_key,
+ cache_key,
+ )
+ cache_payload = ScreenshotCachePayload(scope=cache_scope)
+ cache_payload.pending()
+ try:
+ screenshot_obj.store_cache_payload(
+ next_cache_key,
+ cache_payload,
+ )
+ except ScreenshotCacheError:
+ logger.exception(
+ "Screenshot task preparation failed: %s",
+ next_cache_key,
+ )
+ lock_response = self.response(
+ 503,
+ message=gettext("Screenshot cache is unavailable"),
+ )
+ return lock_response
+
+ try:
+ cache_dashboard_screenshot.delay(
+ username=get_current_user(),
+ guest_token=(
+ g.user.guest_token
+ if get_current_user() and isinstance(g.user,
GuestUser)
+ else None
+ ),
+ dashboard_id=dashboard.id,
+ dashboard_url=dashboard_url,
+ thumb_size=thumb_size,
+ window_size=window_size,
+ cache_key=next_cache_key,
+ # The API has already selected a fresh generation.
+ # Duplicate deliveries should never force a
completed
+ # result to recompute.
+ force=False,
+ )
+ except Exception: # pylint: disable=broad-except
+ screenshot_obj.mark_cache_error_if_incomplete(
+ next_cache_key,
+ cache_scope,
+ )
+ raise
+ try:
+ # Publish only after Celery accepts the task.
Otherwise a
+ # process exit between these operations strands a fresh
+ # Pending generation that no worker can complete.
+ screenshot_obj.set_current_api_generation_cache_key(
+ request_cache_key,
+ next_cache_key,
+ cache_scope,
+ )
+ except ScreenshotCacheError:
+ logger.exception(
+ "Screenshot generation publication failed: %s",
+ next_cache_key,
+ )
+ lock_response = self.response(
+ 503,
+ message=gettext("Screenshot cache is unavailable"),
+ )
+ return lock_response
+ lock_response = build_response(202, next_cache_key,
cache_payload)
+ return lock_response
+ except ReleaseDistributedLockFailedException:
+ if lock_response is not None:
+ logger.warning(
+ "Screenshot request completed but its producer lock
could "
+ "not be released: %s",
+ request_cache_key,
+ exc_info=True,
+ )
+ return lock_response
+ logger.exception(
+ "Could not release screenshot producer lock: %s",
+ request_cache_key,
+ )
+ return self.response(
+ 503,
+ message=gettext("Screenshot request is temporarily
unavailable"),
+ )
+ except LockAlreadyHeldException:
+ try:
+ cache_key, current_payload = get_current_generation()
Review Comment:
Every 50ms retry calls `get_current_generation()`, which pulls the full
payload and base64-decodes the PNG, so waiting out a dead producer's lease is
~600 multi-MB reads plus ~600 lock acquisitions; could the wait loop resolve
only the pointer and fetch the payload once it's about to respond?
--
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]
