chore(hybrid-cloud): Adds silo modes to most unmarked tasks

src/sentry/data_export/tasks.py

@@ -7,6 +7,8 @@
 import celery
 import sentry_sdk
 
+from sentry.silo import SiloMode
+
 # XXX(mdtro): backwards compatible imports for celery 4.4.7, remove after upgrade to 5.2.7
 if celery.version_info >= (5, 2):
     from celery import current_task
@@ -53,6 +55,7 @@
     default_retry_delay=60,
     max_retries=3,
     acks_late=True,
+    silo_mode=SiloMode.REGION,
 )
 def assemble_download(
     data_export_id,
@@ -297,7 +300,12 @@ def store_export_chunk_as_blob(data_export, bytes_written, fileobj, blob_size=DE
         return 0
 
 
-@instrumented_task(name="sentry.data_export.tasks.merge_blobs", queue="data_export", acks_late=True)
+@instrumented_task(
+    name="sentry.data_export.tasks.merge_blobs",
+    queue="data_export",
+    acks_late=True,
+    silo_mode=SiloMode.REGION,
+)
 def merge_export_blobs(data_export_id, **kwargs):
     with sentry_sdk.start_span(op="merge"):
         try:

src/sentry/dynamic_sampling/tasks/boost_low_volume_projects.py

@@ -56,6 +56,7 @@
 )
 from sentry.models import Organization, Project
 from sentry.sentry_metrics import indexer
+from sentry.silo import SiloMode
 from sentry.snuba.dataset import Dataset, EntityKey
 from sentry.snuba.metrics.naming_layer.mri import TransactionMRI
 from sentry.snuba.referrer import Referrer
@@ -71,6 +72,7 @@
     max_retries=5,
     soft_time_limit=2 * 60 * 60,
     time_limit=2 * 60 * 60 + 5,
+    silo_mode=SiloMode.REGION,
 )
 @dynamic_sampling_task_with_context(max_task_execution=MAX_TASK_SECONDS)
 def boost_low_volume_projects(context: TaskContext) -> None:
@@ -91,6 +93,7 @@ def boost_low_volume_projects(context: TaskContext) -> None:
     max_retries=5,
     soft_time_limit=25 * 60,
     time_limit=2 * 60 + 5,
+    silo_mode=SiloMode.REGION,
 )
 @dynamic_sampling_task
 def boost_low_volume_projects_of_org(

src/sentry/dynamic_sampling/tasks/boost_low_volume_transactions.py

@@ -48,6 +48,7 @@
 )
 from sentry.models import Organization
 from sentry.sentry_metrics import indexer
+from sentry.silo import SiloMode
 from sentry.snuba.dataset import Dataset, EntityKey
 from sentry.snuba.metrics.naming_layer.mri import TransactionMRI
 from sentry.snuba.referrer import Referrer
@@ -92,6 +93,7 @@ class ProjectTransactionsTotals(TypedDict, total=True):
     max_retries=5,
     soft_time_limit=2 * 60 * 60,
     time_limit=2 * 60 * 60 + 5,
+    silo_mode=SiloMode.REGION,
 )
 @dynamic_sampling_task_with_context(max_task_execution=MAX_TASK_SECONDS)
 def boost_low_volume_transactions(context: TaskContext) -> None:
@@ -146,6 +148,7 @@ def boost_low_volume_transactions(context: TaskContext) -> None:
     max_retries=5,
     soft_time_limit=25 * 60,
     time_limit=2 * 60 + 5,
+    silo_mode=SiloMode.REGION,
 )
 @dynamic_sampling_task
 def boost_low_volume_transactions_of_project(project_transactions: ProjectTransactions) -> None:

src/sentry/dynamic_sampling/tasks/collect_orgs.py

@@ -6,6 +6,7 @@
 from sentry.dynamic_sampling.tasks.logging import log_task_execution, log_task_timeout
 from sentry.dynamic_sampling.tasks.task_context import TaskContext
 from sentry.dynamic_sampling.tasks.utils import dynamic_sampling_task
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 
 
@@ -16,6 +17,7 @@
     max_retries=5,
     soft_time_limit=2 * 60 * 60,
     time_limit=2 * 60 * 60 + 5,
+    silo_mode=SiloMode.REGION,
 )
 @dynamic_sampling_task
 def collect_orgs() -> None:

src/sentry/dynamic_sampling/tasks/recalibrate_orgs.py

@@ -30,6 +30,7 @@
 )
 from sentry.dynamic_sampling.tasks.task_context import TaskContext
 from sentry.dynamic_sampling.tasks.utils import dynamic_sampling_task_with_context
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 
 # Since we are using a granularity of 60 (minute granularity), we want to have a higher time upper limit for executing
@@ -51,6 +52,7 @@ def __init__(self, org_id, message):
     max_retries=5,
     soft_time_limit=2 * 60 * 60,  # 2hours
     time_limit=2 * 60 * 60 + 5,
+    silo_mode=SiloMode.REGION,
 )
 @dynamic_sampling_task_with_context(max_task_execution=MAX_TASK_SECONDS)
 def recalibrate_orgs(context: TaskContext) -> None:

src/sentry/dynamic_sampling/tasks/sliding_window.py

@@ -41,6 +41,7 @@
 from sentry.dynamic_sampling.tasks.task_context import TaskContext
 from sentry.dynamic_sampling.tasks.utils import dynamic_sampling_task_with_context
 from sentry.sentry_metrics import indexer
+from sentry.silo import SiloMode
 from sentry.snuba.dataset import Dataset, EntityKey
 from sentry.snuba.metrics.naming_layer.mri import TransactionMRI
 from sentry.snuba.referrer import Referrer
@@ -57,6 +58,7 @@
     max_retries=5,
     soft_time_limit=2 * 60 * 60,  # 2 hours
     time_limit=2 * 60 * 60 + 5,
+    silo_mode=SiloMode.REGION,
 )
 @dynamic_sampling_task_with_context(max_task_execution=MAX_TASK_SECONDS)
 def sliding_window(context: TaskContext) -> None:

src/sentry/dynamic_sampling/tasks/sliding_window_org.py

@@ -20,6 +20,7 @@
 )
 from sentry.dynamic_sampling.tasks.task_context import TaskContext
 from sentry.dynamic_sampling.tasks.utils import dynamic_sampling_task_with_context
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 
 
@@ -30,6 +31,7 @@
     max_retries=5,
     soft_time_limit=2 * 60 * 60,  # 2 hours
     time_limit=2 * 60 * 60 + 5,
+    silo_mode=SiloMode.REGION,
 )
 @dynamic_sampling_task_with_context(max_task_execution=MAX_TASK_SECONDS)
 def sliding_window_org(context: TaskContext) -> None:

src/sentry/incidents/tasks.py

@@ -21,6 +21,7 @@
 from sentry.models import Project
 from sentry.services.hybrid_cloud.user import RpcUser
 from sentry.services.hybrid_cloud.user.service import user_service
+from sentry.silo import SiloMode
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import QuerySubscription
 from sentry.snuba.query_subscriptions.consumer import register_subscriber
@@ -36,7 +37,11 @@
 SUBSCRIPTION_METRICS_LOGGER = "subscription_metrics_logger"
 
 
-@instrumented_task(name="sentry.incidents.tasks.send_subscriber_notifications", queue="incidents")
+@instrumented_task(
+    name="sentry.incidents.tasks.send_subscriber_notifications",
+    queue="incidents",
+    silo_mode=SiloMode.REGION,
+)
 def send_subscriber_notifications(activity_id: int) -> None:
     from sentry.incidents.logic import get_incident_subscribers, unsubscribe_from_incident
 
@@ -169,6 +174,7 @@ def handle_snuba_query_update(
     queue="incidents",
     default_retry_delay=60,
     max_retries=5,
+    silo_mode=SiloMode.REGION,
 )
 def handle_trigger_action(
     action_id: int,
@@ -215,6 +221,7 @@ def handle_trigger_action(
     queue="incidents",
     default_retry_delay=60,
     max_retries=2,
+    silo_mode=SiloMode.REGION,
 )
 def auto_resolve_snapshot_incidents(alert_rule_id: int, **kwargs: Any) -> None:
     from sentry.incidents.logic import update_incident_status

src/sentry/issues/forecasts.py

@@ -15,6 +15,7 @@
 from sentry.issues.escalating_group_forecast import EscalatingGroupForecast
 from sentry.issues.escalating_issues_alg import generate_issue_forecast, standard_version
 from sentry.models import Group
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 
 logger = logging.getLogger(__name__)
@@ -65,6 +66,7 @@ def generate_and_save_forecasts(groups: Sequence[Group]) -> None:
 @instrumented_task(
     name="sentry.tasks.weekly_escalating_forecast.generate_and_save_missing_forecasts",
     queue="weekly_escalating_forecast",
+    silo_mode=SiloMode.REGION,
 )
 def generate_and_save_missing_forecasts(group_id: int) -> None:
     """

src/sentry/monitors/tasks.py

@@ -3,6 +3,7 @@
 from django.utils import timezone
 
 from sentry.constants import ObjectStatus
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 from sentry.utils import metrics
 
@@ -40,7 +41,12 @@
 SUBTITLE_DATETIME_FORMAT = "%b %d, %I:%M %p"
 
 
-@instrumented_task(name="sentry.monitors.tasks.check_missing", time_limit=15, soft_time_limit=10)
+@instrumented_task(
+    name="sentry.monitors.tasks.check_missing",
+    time_limit=15,
+    soft_time_limit=10,
+    silo_mode=SiloMode.REGION,
+)
 def check_missing(current_datetime=None):
     if current_datetime is None:
         current_datetime = timezone.now()
@@ -106,7 +112,12 @@ def check_missing(current_datetime=None):
             logger.exception("Exception in check_monitors - mark missed")
 
 
-@instrumented_task(name="sentry.monitors.tasks.check_timeout", time_limit=15, soft_time_limit=10)
+@instrumented_task(
+    name="sentry.monitors.tasks.check_timeout",
+    time_limit=15,
+    soft_time_limit=10,
+    silo_mode=SiloMode.REGION,
+)
 def check_timeout(current_datetime=None):
     if current_datetime is None:
         current_datetime = timezone.now()

src/sentry/profiles/task.py

@@ -21,6 +21,7 @@
 from sentry.profiles.java import deobfuscate_signature
 from sentry.profiles.utils import get_from_profiling_service
 from sentry.signals import first_profile_received
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 from sentry.utils import json, metrics
 from sentry.utils.outcomes import Outcome, track_outcome
@@ -46,6 +47,7 @@ class VroomTimeout(Exception):
     acks_late=True,
     task_time_limit=60,
     task_acks_on_failure_or_timeout=False,
+    silo_mode=SiloMode.REGION,
 )
 def process_profile_task(
     profile: Optional[Profile] = None,

src/sentry/replays/tasks.py

@@ -9,6 +9,7 @@
 from sentry.replays.lib.storage import FilestoreBlob, StorageBlob
 from sentry.replays.models import ReplayRecordingSegment
 from sentry.replays.usecases.reader import fetch_segments_metadata
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 from sentry.utils import json
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
@@ -22,6 +23,7 @@
     queue="replays.delete_replay",
     default_retry_delay=5,
     max_retries=5,
+    silo_mode=SiloMode.REGION,
 )
 def delete_recording_segments(project_id: int, replay_id: str, **kwargs: Any) -> None:
     """Asynchronously delete a replay."""

src/sentry/snuba/tasks.py

@@ -32,6 +32,8 @@
 SUBSCRIPTION_STATUS_MAX_AGE = timedelta(minutes=10)
 
 
+# TODO(hybrid-cloud): Mark this as region silo only once testing/decorator
+#  interaction is cleaned up
 @instrumented_task(
     name="sentry.snuba.tasks.create_subscription_in_snuba",
     queue="subscriptions",

src/sentry/tasks/activity.py

@@ -1,5 +1,6 @@
 import logging
 
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 from sentry.utils.safe import safe_execute
 from sentry.utils.sdk import bind_organization_context
@@ -27,7 +28,9 @@ def get_activity_notifiers(project):
 
 
 @instrumented_task(
-    name="sentry.tasks.activity.send_activity_notifications", queue="activity.notify"
+    name="sentry.tasks.activity.send_activity_notifications",
+    queue="activity.notify",
+    silo_mode=SiloMode.REGION,
 )
 def send_activity_notifications(activity_id):
     from sentry.models import Activity, Organization

src/sentry/tasks/assemble.py

@@ -34,6 +34,7 @@
     ReleaseArtifactBundle,
 )
 from sentry.models.releasefile import ReleaseArchive, update_artifact_index
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 from sentry.utils import metrics
 from sentry.utils.db import atomic_transaction
@@ -191,7 +192,11 @@ def set_assemble_status(task, scope, checksum, state, detail=None):
     default_cache.set(cache_key, (state, detail), 600)
 
 
-@instrumented_task(name="sentry.tasks.assemble.assemble_dif", queue="assemble")
+@instrumented_task(
+    name="sentry.tasks.assemble.assemble_dif",
+    queue="assemble",
+    silo_mode=SiloMode.REGION,
+)
 def assemble_dif(project_id, name, checksum, chunks, debug_id=None, **kwargs):
     """
     Assembles uploaded chunks into a ``ProjectDebugFile``.
@@ -802,7 +807,11 @@ def prepare_post_assembler(
         )
 
 
-@instrumented_task(name="sentry.tasks.assemble.assemble_artifacts", queue="assemble")
+@instrumented_task(
+    name="sentry.tasks.assemble.assemble_artifacts",
+    queue="assemble",
+    silo_mode=SiloMode.REGION,
+)
 def assemble_artifacts(
     org_id,
     version,

src/sentry/tasks/auth.py

@@ -13,6 +13,7 @@
 from sentry.services.hybrid_cloud.organization import organization_service
 from sentry.services.hybrid_cloud.user import RpcUser
 from sentry.services.hybrid_cloud.user.service import user_service
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task, retry
 from sentry.utils.audit import create_audit_entry_from_user
 from sentry.utils.email import MessageBuilder
@@ -21,6 +22,7 @@
 logger = logging.getLogger("sentry.auth")
 
 
+# TODO(hybrid-cloud): Remove cross-silo DB accesses, or review control silo usages
 @instrumented_task(name="sentry.tasks.send_sso_link_emails", queue="auth")
 def email_missing_links(org_id: int, actor_id: int, provider_key: str, **kwargs):
     try:
@@ -42,6 +44,7 @@ def email_missing_links(org_id: int, actor_id: int, provider_key: str, **kwargs)
         member.send_sso_link_email(user.id, provider)
 
 
+# TODO(hybrid-cloud): Remove cross-silo DB accesses, or review control silo usages
 @instrumented_task(name="sentry.tasks.email_unlink_notifications", queue="auth")
 def email_unlink_notifications(org_id: int, actor_id: int, provider_key: str):
     try:
@@ -164,6 +167,7 @@ def call_to_action(self, org: Organization, user: RpcUser, member: OrganizationM
     queue="auth",
     default_retry_delay=60 * 5,
     max_retries=5,
+    silo_mode=SiloMode.REGION,
 )
 @retry
 def remove_2fa_non_compliant_members(org_id, actor_id=None, actor_key_id=None, ip_address=None):
@@ -220,6 +224,7 @@ def call_to_action(self, org: Organization, user: RpcUser, member: OrganizationM
     queue="auth",
     default_retry_delay=60 * 5,
     max_retries=5,
+    silo_mode=SiloMode.REGION,
 )
 @retry
 def remove_email_verification_non_compliant_members(

src/sentry/tasks/auto_enable_codecov.py

@@ -3,6 +3,7 @@
 from sentry import audit_log, features
 from sentry.integrations.utils.codecov import has_codecov_integration
 from sentry.models.organization import Organization, OrganizationStatus
+from sentry.silo import SiloMode
 from sentry.tasks.base import instrumented_task
 from sentry.utils.audit import create_system_audit_entry
 from sentry.utils.query import RangeQuerySetWrapper
@@ -14,6 +15,7 @@
     name="sentry.tasks.auto_enable_codecov.enable_for_org",
     queue="auto_enable_codecov",
     max_retries=0,
+    silo_mode=SiloMode.REGION,
 )
 def enable_for_org(dry_run: bool = False) -> None:
     """