codeant-ai-for-open-source[bot] commented on code in PR #42089:
URL: https://github.com/apache/superset/pull/42089#discussion_r3694438155
##########
docs/admin_docs/configuration/alerts-reports.mdx:
##########
@@ -81,6 +81,23 @@ SLACK_CACHE_TIMEOUT = int(timedelta(days=2).total_seconds())
SLACK_API_RATE_LIMIT_RETRY_COUNT = 5
```
+When the cache backend is `SupersetMetastoreCache`, report execution does not
+write channel listings into the cache because that backend commits the report's
+database session. Schedule the dedicated warm-up task so cache misses are
+repopulated outside report transactions:
+
+```python
+from celery.schedules import crontab
+
+class CeleryConfig:
+ beat_schedule = {
+ "slack.cache_channels": {
+ "task": "slack.cache_channels",
+ "schedule": crontab(minute="0", hour="*"),
+ },
Review Comment:
**Suggestion:** Copying this example into `superset_config.py` replaces
Superset's entire `beat_schedule`, removing the built-in `reports.scheduler`,
`reports.prune_log`, version-history, and deletion-retention entries. This can
silently stop report execution and maintenance tasks. Show how to extend the
existing schedule while preserving the default entries instead of assigning a
new schedule containing only the Slack task. [logic error]
<details>
<summary><b>Severity Level:</b> Critical 🚨</summary>
```mdx
- ❌ Scheduled alerts and reports stop triggering.
- ❌ Version-history retention task stops running.
- ❌ Soft-deletion retention task stops running.
- ⚠️ Slack cache warm-up remains the only scheduled entry.
```
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=5cbc942a1cfd40bba46f9863e621b54a&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset)
[](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=5cbc942a1cfd40bba46f9863e621b54a&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset)
*(Use Cmd/Ctrl + Click for best experience)*
<details>
<summary><b>Prompt for AI Agent 🤖 </b></summary>
```mdx
This is a comment left during a code review.
**Path:** docs/admin_docs/configuration/alerts-reports.mdx
**Line:** 92:97
**Comment:**
*Logic Error: Copying this example into `superset_config.py` replaces
Superset's entire `beat_schedule`, removing the built-in `reports.scheduler`,
`reports.prune_log`, version-history, and deletion-retention entries. This can
silently stop report execution and maintenance tasks. Show how to extend the
existing schedule while preserving the default entries instead of assigning a
new schedule containing only the Slack task.
Validate the correctness of the flagged issue. If correct, How can I resolve
this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask
user if the user wants to fix the rest of the comments as well. if said yes,
then fetch all the comments validate the correctness and implement a minimal fix
```
</details>
<a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42089&comment_hash=f8a62c5afd3574a505686c362292aa424ee325f62754ebfde679ed1afc302df1&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42089&comment_hash=f8a62c5afd3574a505686c362292aa424ee325f62754ebfde679ed1afc302df1&reaction=dislike'>👎</a>
##########
superset/commands/report/slack_upgrade.py:
##########
@@ -0,0 +1,201 @@
+# 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.
+
+import logging
+from collections.abc import Callable
+from uuid import UUID
+
+from flask import current_app as app
+
+from superset.commands.exceptions import UpdateFailedError
+from superset.reports.models import (
+ ReportRecipients,
+ ReportRecipientType,
+ ReportSchedule,
+)
+from superset.reports.notifications.base import BaseNotification,
NotificationContent
+from superset.reports.notifications.exceptions import (
+ NotificationParamException,
+ SlackV1NotificationError,
+)
+from superset.reports.notifications.slack import (
+ SLACK_V1_FILE_UPLOAD_MESSAGE,
+ SlackNotification,
+)
+from superset.reports.notifications.slack_channel_resolver import (
+ resolve_slack_channel_ids,
+)
+from superset.utils import json
+from superset.utils.decorators import record_statsd_gauge_failure
+from superset.utils.slack import (
+ NO_SLACK_RECIPIENTS_MESSAGE,
+ parse_slack_recipient_targets,
+ SlackChannelListingClientError,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class SlackV1UpgradeCoordinator:
+ """Coordinate one atomic Slack v1 upgrade and its per-recipient
fallbacks."""
+
+ def __init__(
+ self,
+ report_schedule: ReportSchedule,
+ execution_id: UUID,
+ execution_warnings: list[str],
+ ) -> None:
+ self._report_schedule = report_schedule
+ self._execution_id = execution_id
+ self._execution_warnings = execution_warnings
+ self.reset()
+
+ def reset(self) -> None:
+ """Reset execution-scoped upgrade and fallback state."""
+ self._upgrade_error: NotificationParamException | UpdateFailedError |
None = (
+ None
+ )
+ self._fallback_recorded = False
+
+ def update_recipients(self) -> None:
+ """Resolve and atomically convert every Slack v1 recipient to Slack
v2."""
+ pending: list[tuple[ReportRecipients, list[str]]] = []
+ try:
+ for recipient in self._report_schedule.recipients:
+ if recipient.type != ReportRecipientType.SLACK:
+ continue
+ try:
+ slack_recipients =
json.loads(recipient.recipient_config_json)
+ except (TypeError, ValueError) as ex:
+ raise NotificationParamException(
+ "Invalid Slack recipient configuration"
+ ) from ex
+ target = (
+ slack_recipients.get("target")
+ if isinstance(slack_recipients, dict)
+ else None
+ )
+ if not isinstance(target, str):
+ raise
NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+ channels = parse_slack_recipient_targets(target.replace("#",
""))
+ if not channels:
+ raise
NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+ pending.append((recipient, channels))
+
+ all_targets = list(
+ dict.fromkeys(
+ channel for _, channels in pending for channel in channels
+ )
+ )
+ channel_ids = resolve_slack_channel_ids(all_targets) if
all_targets else {}
+ resolved = [
+ (
+ recipient,
+ json.dumps(
+ {
+ "target": ",".join(
+ channel_ids[channel] for channel in channels
+ )
+ }
+ ),
+ )
+ for recipient, channels in pending
+ ]
+ except (NotificationParamException, SlackChannelListingClientError) as
ex:
+ message = f"Failed to update slack recipients to v2: {ex}"
+ logger.warning(message)
+ raise NotificationParamException(message) from ex
+ except Exception as ex:
+ message = f"Failed to update slack recipients to v2: {ex}"
+ logger.exception(message)
+ raise UpdateFailedError(message) from ex
+
+ for recipient, recipient_config_json in resolved:
+ recipient.type = ReportRecipientType.SLACKV2
+ recipient.recipient_config_json = recipient_config_json
Review Comment:
**Suggestion:** Persisting the recipient conversion before the reconstructed
Slack v2 notification succeeds leaves the schedule permanently upgraded when
that send fails. The surrounding report-state logging commits the session, so a
failed v2 delivery can still commit these mutations; subsequent executions then
bypass the Slack v1 fallback and retry the failed v2 path. Roll back the
recipient changes when the upgraded send fails, or only persist the upgrade
after successful delivery. [stale reference]
<details>
<summary><b>Severity Level:</b> Major ⚠️</summary>
```mdx
- ❌ Failed v2 delivery persists an incomplete migration.
- ❌ Later report executions bypass Slack v1 fallback.
- ⚠️ Saved private-channel recipients can remain undeliverable.
```
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=1d8a7d69cd1d42fe94661124f1f28c59&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset)
[](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=1d8a7d69cd1d42fe94661124f1f28c59&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset)
*(Use Cmd/Ctrl + Click for best experience)*
<details>
<summary><b>Prompt for AI Agent 🤖 </b></summary>
```mdx
This is a comment left during a code review.
**Path:** superset/commands/report/slack_upgrade.py
**Line:** 127:129
**Comment:**
*Stale Reference: Persisting the recipient conversion before the
reconstructed Slack v2 notification succeeds leaves the schedule permanently
upgraded when that send fails. The surrounding report-state logging commits the
session, so a failed v2 delivery can still commit these mutations; subsequent
executions then bypass the Slack v1 fallback and retry the failed v2 path. Roll
back the recipient changes when the upgraded send fails, or only persist the
upgrade after successful delivery.
Validate the correctness of the flagged issue. If correct, How can I resolve
this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask
user if the user wants to fix the rest of the comments as well. if said yes,
then fetch all the comments validate the correctness and implement a minimal fix
```
</details>
<a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42089&comment_hash=6e30e0d6bce6c1c4ce10ef29c89c3a49a5484817720cbab9cdf10277ddce6be2&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42089&comment_hash=6e30e0d6bce6c1c4ce10ef29c89c3a49a5484817720cbab9cdf10277ddce6be2&reaction=dislike'>👎</a>
##########
superset/utils/slack.py:
##########
@@ -194,65 +401,226 @@ def _get_channels(
raise
-def get_channels_with_search(
+def _filter_slack_channels(
+ channels: list[SlackChannel],
+ *,
+ search_string: str,
+ types: Optional[list[SlackChannelTypes]],
+ exact_match: bool,
+) -> list[SlackChannel]:
+ """Filter a complete Slack channel listing by type and target."""
+ if types and len(types) != len(SlackChannelTypes):
+ conditions: list[Callable[[SlackChannel], bool]] = []
+ if SlackChannelTypes.PUBLIC in types:
+ conditions.append(lambda channel: not channel["is_private"])
+ if SlackChannelTypes.PRIVATE in types:
+ conditions.append(lambda channel: channel["is_private"])
+
+ channels = [
+ channel for channel in channels if any(cond(channel) for cond in
conditions)
+ ]
+
+ if not search_string:
+ return channels
+
+ search_array = recipients_string_to_list(search_string)
+ return [
+ channel
+ for channel in channels
+ if any(
+ (
+ search.casefold() == channel["name"].casefold()
+ or search.casefold() == channel["id"].casefold()
+ if exact_match
+ else (
+ search.casefold() in channel["name"].casefold()
+ or search.casefold() in channel["id"].casefold()
+ )
+ )
+ for search in search_array
+ )
+ ]
+
+
+def _get_channels_with_search(
search_string: str = "",
types: Optional[list[SlackChannelTypes]] = None,
exact_match: bool = False,
force: bool = False,
-) -> list[SlackChannelSchema]:
+ cache: bool = True,
+ *,
+ return_cache_status: bool = False,
+) -> tuple[list[SlackChannel], bool]:
"""
The slack api is paginated but does not include search, so we need to fetch
all channels and filter them ourselves
This will search by slack name or id
"""
+ used_cache = False
+ cache_timeout = app.config["SLACK_CACHE_TIMEOUT"]
+ cache_enabled = cache and cache_timeout != CACHE_DISABLED_TIMEOUT
try:
- channels = get_channels(
- force=force,
- cache_timeout=app.config["SLACK_CACHE_TIMEOUT"],
- )
+ if return_cache_status and cache_enabled and not force:
+ channels, used_cache = _get_channels_with_cache_status()
+ else:
+ channels = get_channels(
+ force=force,
+ cache=cache_enabled,
+ cache_timeout=cache_timeout,
+ )
except SlackApiError as ex:
- # Check if it's a rate limit error
- status_code = getattr(ex.response, "status_code", None)
- if status_code == 429:
- raise SupersetException(
- f"Slack API rate limit exceeded: {ex}. "
- "For large workspaces, consider increasing "
- "SLACK_API_RATE_LIMIT_RETRY_COUNT"
- ) from ex
- raise SupersetException(f"Failed to list channels: {ex}") from ex
- except SlackClientError as ex:
- raise SupersetException(f"Failed to list channels: {ex}") from ex
+ error_code = get_slack_api_error_code(ex)
+ error_class = (
+ SlackChannelListingError
+ if is_transient_slack_api_error(ex, error_code)
+ else SlackChannelListingClientError
+ )
+ message = f"Failed to list channels: {ex}"
+ if get_slack_api_status_code(ex) == 429:
+ message = (
+ f"Slack API rate limit exceeded: {ex}. For large workspaces, "
+ "consider increasing SLACK_API_RATE_LIMIT_RETRY_COUNT"
+ )
+ raise error_class(message) from ex
+ except SLACK_TRANSIENT_TRANSPORT_ERRORS as ex:
+ error_class = (
+ SlackChannelListingError
+ if is_transient_slack_transport_error(ex)
+ else SlackChannelListingClientError
+ )
+ raise error_class(f"Failed to list channels: {ex}") from ex
+ except (SlackSDKClientError, SlackClientError) as ex:
+ raise SlackChannelListingClientError(f"Failed to list channels: {ex}")
from ex
+
+ channels = _filter_slack_channels(
+ channels,
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ )
+ return channels, used_cache
- if types and not len(types) == len(SlackChannelTypes):
- conditions: list[Callable[[SlackChannelSchema], bool]] = []
- if SlackChannelTypes.PUBLIC in types:
- conditions.append(lambda channel: not channel["is_private"])
- if SlackChannelTypes.PRIVATE in types:
- conditions.append(lambda channel: channel["is_private"])
- channels = [
- channel for channel in channels if any(cond(channel) for cond in
conditions)
- ]
+def get_channels_with_search(
+ search_string: str = "",
+ types: Optional[list[SlackChannelTypes]] = None,
+ exact_match: bool = False,
+ force: bool = False,
+ cache: bool = True,
+) -> list[SlackChannel]:
+ """Fetch and filter Slack channels without exposing cache provenance."""
+ channels, _ = _get_channels_with_search(
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ force=force,
+ cache=cache,
+ )
+ return channels
- # The search string can be multiple channels separated by commas
- if search_string:
- search_array = recipients_string_to_list(search_string)
- channels = [
- channel
- for channel in channels
- if any(
- (
- search.lower() == channel["name"].lower()
- or search.lower() == channel["id"].lower()
- if exact_match
- else (
- search.lower() in channel["name"].lower()
- or search.lower() in channel["id"].lower()
- )
- )
- for search in search_array
+
+def get_channels_with_search_and_cache_status(
+ search_string: str = "",
+ types: Optional[list[SlackChannelTypes]] = None,
+ exact_match: bool = False,
+) -> tuple[list[SlackChannel], bool]:
+ """Fetch filtered Slack channels and report whether the listing was
cached."""
+ return _get_channels_with_search(
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ return_cache_status=True,
+ )
+
+
+def refresh_cached_slack_channels_with_search(
+ search_string: str = "",
+ types: Optional[list[SlackChannelTypes]] = None,
+ exact_match: bool = False,
+) -> list[SlackChannel]:
+ """Refresh stale channels with a best-effort cache-backend cooldown.
+
+ External cache backends record a cooldown only after the refreshed listing
+ is stored successfully. Disabled and metastore-backed caches use an
uncached
+ request without a cooldown because metastore writes commit the report
+ transaction. Concurrent workers can still refresh in parallel when the
+ backend cannot provide transaction-safe coordination.
+ """
+ team_id = get_team_id()
+ cache_key = _get_slack_channels_cache_key(team_id)
+ cooldown_key = f"{cache_key}_refresh_cooldown"
+ cache_timeout = app.config["SLACK_CACHE_TIMEOUT"]
+
+ if (
+ _slack_channel_cache_uses_report_session()
+ or cache_timeout == CACHE_DISABLED_TIMEOUT
+ ):
+ return get_channels_with_search(
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ force=True,
+ cache=False,
+ )
+
+ try:
+ refresh_is_recent = cache_manager.cache.get(cooldown_key) is not None
+ except Exception: # pylint: disable=broad-exception-caught
+ refresh_is_recent = False
+ logger.warning(
+ "Could not read Slack channel refresh cooldown; refreshing from
Slack",
+ exc_info=True,
+ )
+
+ if refresh_is_recent:
+ channels, _ = _get_channels_with_cache_status()
+ return _filter_slack_channels(
+ channels,
+ search_string=search_string,
+ types=types,
+ exact_match=exact_match,
+ )
+
+ refreshed_channels = get_channels_with_search(
Review Comment:
**Suggestion:** The refresh cooldown check is not an atomic claim:
concurrent workers can all observe no cooldown, fetch the complete Slack
channel listing, and only then write the cooldown. During simultaneous report
upgrades this multiplies `conversations.list` requests and can trigger Slack
rate limits. Use an atomic cache add/lock before refreshing, or otherwise
coordinate refresh ownership across workers. [race condition]
<details>
<summary><b>Severity Level:</b> Major ⚠️</summary>
```mdx
- ⚠️ Concurrent report upgrades duplicate full Slack channel listings.
- ⚠️ Large workspaces can consume Slack rate-limit capacity.
- ⚠️ Rate limiting can delay or fail recipient upgrades.
```
</details>
[](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=b2e3ecd2a97b4f4fbf192fe49802c866&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset)
[](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=b2e3ecd2a97b4f4fbf192fe49802c866&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset)
*(Use Cmd/Ctrl + Click for best experience)*
<details>
<summary><b>Prompt for AI Agent 🤖 </b></summary>
```mdx
This is a comment left during a code review.
**Path:** superset/utils/slack.py
**Line:** 566:584
**Comment:**
*Race Condition: The refresh cooldown check is not an atomic claim:
concurrent workers can all observe no cooldown, fetch the complete Slack
channel listing, and only then write the cooldown. During simultaneous report
upgrades this multiplies `conversations.list` requests and can trigger Slack
rate limits. Use an atomic cache add/lock before refreshing, or otherwise
coordinate refresh ownership across workers.
Validate the correctness of the flagged issue. If correct, How can I resolve
this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask
user if the user wants to fix the rest of the comments as well. if said yes,
then fetch all the comments validate the correctness and implement a minimal fix
```
</details>
<a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42089&comment_hash=353600ad08a2d81c710a4da7f2eb57ff461a65631eb60605be249f2c9c976535&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F42089&comment_hash=353600ad08a2d81c710a4da7f2eb57ff461a65631eb60605be249f2c9c976535&reaction=dislike'>👎</a>
--
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]
