sadpandajoe commented on code in PR #43316: URL: https://github.com/apache/superset/pull/43316#discussion_r3826073974
########## superset/coordination/base.py: ########## @@ -0,0 +1,386 @@ +# 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. +"""Coordination service implementation. + +See :mod:`superset.coordination` for the package overview. +""" + +from __future__ import annotations + +import logging +import threading +import time +from typing import Any, Callable, TYPE_CHECKING, TypeVar + +from superset.coordination.exceptions import CoordinationBackendUnavailableError +from superset.coordination.types import SignalListener +from superset.coordination.utils import close_pubsub + +if TYPE_CHECKING: + from superset.coordination.types import CoordinationBackend + +logger = logging.getLogger(__name__) + +T = TypeVar("T") + +# Poll cadence for the pub/sub wait loop: how long each ``get_message`` blocks +# before the loop re-checks the predicate, the timeout, and the stop flag. Keeps +# stop latency and missed-message recovery bounded to ~1s. +_PUBSUB_TICK_SECONDS = 1.0 + + +class CoordinationService: + """Single entry point for the Valkey/Redis coordination primitives. + + Two layers of API: + + - **Raw primitives** — ``publish``, ``get`` / ``set`` / ``delete``, + ``stream_add`` / ``stream_range``. These are backend-only and have no fallback: + they raise :class:`CoordinationBackendUnavailableError` when no backend is + configured, rather than silently doing nothing. + - **Higher-level await/notify** — ``wait_for_signal`` (blocking) and + ``listen_for_signal`` (background). These combine a pub/sub channel with a + caller-supplied predicate: + when a backend is defined they wake promptly on a published message, and either + way they fall back to polling the predicate. This keeps the pub/sub-vs-poll + boilerplate in one place; callers just supply a channel and a check. + + All methods are class-level: the service is app-global and resolves its backend + from the shared coordination connection on each call. + + Distributed locking is *not* exposed here: it has its own user-facing interface + (:class:`~superset.distributed_lock.DistributedLock`) that uses this service's + backend when one is defined and falls back to a database-backed lock otherwise. + """ + + _legacy_backend: "CoordinationBackend | None" = None + _legacy_warning_emitted: bool = False + + @classmethod + def get_backend(cls) -> "CoordinationBackend | None": + """Resolve the coordination backend. + + Prefers ``DISTRIBUTED_COORDINATION_CONFIG`` (via the cache manager). Falls + back to the deprecated ``GLOBAL_ASYNC_QUERIES_CACHE_BACKEND`` when only that + is configured, emitting a one-time deprecation warning. Returns ``None`` when + neither is configured. + """ + from superset.extensions import cache_manager + + if (backend := cache_manager.distributed_coordination) is not None: Review Comment: This resolver now changes the backend for callers that are not GAQ: with only the legacy GAQ setting, locks and GTF move from the metastore to Redis; with both settings, GAQ moves to the coordination Redis. During a rolling upgrade those callers can therefore use different stores and lose mutual exclusion or event delivery. Should the legacy/new selection stay scoped to the GAQ call sites until every consumer is migrated? ########## superset/coordination/base.py: ########## @@ -0,0 +1,386 @@ +# 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. +"""Coordination service implementation. + +See :mod:`superset.coordination` for the package overview. +""" + +from __future__ import annotations + +import logging +import threading +import time +from typing import Any, Callable, TYPE_CHECKING, TypeVar + +from superset.coordination.exceptions import CoordinationBackendUnavailableError +from superset.coordination.types import SignalListener +from superset.coordination.utils import close_pubsub + +if TYPE_CHECKING: + from superset.coordination.types import CoordinationBackend + +logger = logging.getLogger(__name__) + +T = TypeVar("T") + +# Poll cadence for the pub/sub wait loop: how long each ``get_message`` blocks +# before the loop re-checks the predicate, the timeout, and the stop flag. Keeps +# stop latency and missed-message recovery bounded to ~1s. +_PUBSUB_TICK_SECONDS = 1.0 + + +class CoordinationService: + """Single entry point for the Valkey/Redis coordination primitives. + + Two layers of API: + + - **Raw primitives** — ``publish``, ``get`` / ``set`` / ``delete``, + ``stream_add`` / ``stream_range``. These are backend-only and have no fallback: + they raise :class:`CoordinationBackendUnavailableError` when no backend is + configured, rather than silently doing nothing. + - **Higher-level await/notify** — ``wait_for_signal`` (blocking) and + ``listen_for_signal`` (background). These combine a pub/sub channel with a + caller-supplied predicate: + when a backend is defined they wake promptly on a published message, and either + way they fall back to polling the predicate. This keeps the pub/sub-vs-poll + boilerplate in one place; callers just supply a channel and a check. + + All methods are class-level: the service is app-global and resolves its backend + from the shared coordination connection on each call. + + Distributed locking is *not* exposed here: it has its own user-facing interface + (:class:`~superset.distributed_lock.DistributedLock`) that uses this service's + backend when one is defined and falls back to a database-backed lock otherwise. + """ + + _legacy_backend: "CoordinationBackend | None" = None + _legacy_warning_emitted: bool = False + + @classmethod + def get_backend(cls) -> "CoordinationBackend | None": + """Resolve the coordination backend. + + Prefers ``DISTRIBUTED_COORDINATION_CONFIG`` (via the cache manager). Falls + back to the deprecated ``GLOBAL_ASYNC_QUERIES_CACHE_BACKEND`` when only that + is configured, emitting a one-time deprecation warning. Returns ``None`` when + neither is configured. + """ + from superset.extensions import cache_manager + + if (backend := cache_manager.distributed_coordination) is not None: + return backend + return cls._get_legacy_backend() + + @classmethod + def _get_legacy_backend(cls) -> "CoordinationBackend | None": + if cls._legacy_backend is not None: + return cls._legacy_backend + + from flask import current_app + + from superset import is_feature_enabled + + # GLOBAL_ASYNC_QUERIES_CACHE_BACKEND ships a populated default, so its mere + # presence is not operator intent — it only signals a coordination backend + # when Global Async Queries is actually enabled. Without this gate every + # deployment (and all lock/GTF callers) would treat the default as a live + # Redis backend and try to connect. The legacy bridge exists solely to keep + # GAQ working during the deprecation window. + if not is_feature_enabled("GLOBAL_ASYNC_QUERIES"): + return None + + if not current_app.config.get("GLOBAL_ASYNC_QUERIES_CACHE_BACKEND", {}).get( + "CACHE_TYPE" + ): + return None + + if not cls._legacy_warning_emitted: + logger.warning( + "GLOBAL_ASYNC_QUERIES_CACHE_BACKEND is deprecated and will be " + "removed in Superset 8.0; configure DISTRIBUTED_COORDINATION_CONFIG " + "instead so a single connection powers distributed locks, pub/sub, " + "and streams." + ) + cls._legacy_warning_emitted = True + + from superset.async_events.async_query_manager import get_cache_backend + + cls._legacy_backend = get_cache_backend(current_app.config) + return cls._legacy_backend + + @classmethod + def is_backend_defined(cls) -> bool: + """Whether a coordination backend is defined. + + Some operations require the Valkey/Redis backend + (``DISTRIBUTED_COORDINATION_CONFIG``) to be configured; those that do note it + on their own docstring. Best-effort callers should branch on this before + invoking a backend-dependent operation instead of catching + :class:`CoordinationBackendUnavailableError`. + """ + return cls.get_backend() is not None + + @classmethod + def _require_backend(cls) -> "CoordinationBackend": + """Return the backend or raise if none is configured. + + Used by the backend-only primitives (pub/sub publish, key/value, streams) + so a missing backend fails loudly instead of silently no-op'ing. + """ + backend = cls.get_backend() + if backend is None: + raise CoordinationBackendUnavailableError( + "No coordination backend configured; set " + "DISTRIBUTED_COORDINATION_CONFIG to enable key/value and stream " + "operations." + ) + return backend + + # -- Pub/Sub ------------------------------------------------------------- + + @classmethod + def publish(cls, channel: str, message: str) -> int: + """Publish a message to a channel; returns the subscriber count. + + Only publishing is offered here — subscribing needs the native connection + (a long-lived subscription with its own receive loop), so consumers that + subscribe should obtain it via :meth:`get_backend`. + + :raises CoordinationBackendUnavailableError: if no backend is configured. + """ + return cls._require_backend().publish(channel, message) + + # -- Key/Value ----------------------------------------------------------- + + @classmethod + def get_value(cls, key: str) -> Any: + """Return the raw (bytes) value at ``key``, or ``None`` if absent. + + :raises CoordinationBackendUnavailableError: if no backend is configured. + """ + return cls._require_backend().get(key) + + @classmethod + def set_value( + cls, + key: str, + value: Any, + ttl: int | None = None, + if_absent: bool = False, + if_present: bool = False, + ) -> bool | None: + """Store ``value`` at ``key``. + + :param ttl: optional expiry, in seconds. + :param if_absent: only set if the key does not already exist. + :param if_present: only set if the key already exists. + :returns: ``True`` on success, or ``None`` when an ``if_absent`` / + ``if_present`` condition prevented the write. + :raises CoordinationBackendUnavailableError: if no backend is configured. + """ + return cls._require_backend().set( + key, value, ex=ttl, nx=if_absent, xx=if_present + ) + + @classmethod + def delete_value(cls, *keys: str) -> int: + """Delete one or more keys; returns the number deleted. + + :raises CoordinationBackendUnavailableError: if no backend is configured. + """ + return cls._require_backend().delete(*keys) + + # -- Streams ------------------------------------------------------------- + + @classmethod + def stream_add( + cls, + stream: str, + data: dict[str, Any], + event_id: str = "*", + max_len: int | None = None, + ) -> str: + """Append an event to a stream; returns the generated event id. + + :raises CoordinationBackendUnavailableError: if no backend is configured. + """ + return cls._require_backend().xadd(stream, data, event_id, max_len) + + @classmethod + def stream_range( + cls, + stream: str, + start: str = "-", + end: str = "+", + count: int | None = None, + ) -> list[Any]: + """Read a range of events from a stream. + + :raises CoordinationBackendUnavailableError: if no backend is configured. + """ + return cls._require_backend().xrange(stream, start, end, count) + + # -- Await / notify ------------------------------------------------------ + + @classmethod + def wait_for_signal( + cls, + channel: str, + check: Callable[[], T | None], + *, + timeout: float | None = None, + poll_interval: float = 1.0, + ) -> T: + """Block until ``check()`` returns a non-``None`` value; return that value. + + ``check`` is the source of truth (typically a metastore read). When a + coordination backend is defined, this subscribes to ``channel`` and re-runs + ``check`` promptly whenever a message is published; otherwise it polls + ``check`` every ``poll_interval`` seconds. ``check`` is also re-evaluated on + every tick even in pub/sub mode, so a signal published before the subscription + (or a dropped message) is still caught. + + :param channel: pub/sub channel that peers publish to when the awaited state + is reached (used only as a low-latency wake-up; correctness relies on + ``check``). + :param check: returns a truthy result once the wait is satisfied, else + ``None``. + :param timeout: max seconds to wait; ``None`` waits indefinitely. + :param poll_interval: poll cadence when no backend is defined. + :raises TimeoutError: if ``timeout`` elapses before ``check`` is satisfied. + """ + deadline = None if timeout is None else time.monotonic() + timeout + backend = cls.get_backend() + pubsub = backend.pubsub() if backend is not None else None + try: + if pubsub is not None: Review Comment: A completed task should not require Redis to be reachable: this subscribes before the first predicate check, so `wait_for_completion` now raises on a Redis outage even when the task is already terminal. Could the initial `check()` run before opening the pub/sub subscription? -- 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]
