feat(hybridcloud) First pass at a siloed devserver (#49956)

With foreign keys mostly broken, we can start building the siloed
development environment. While this doesn't include all the desired
functionality it has working webservers + and a functioning RPC bridge.

### What's included

* Webserver for control + region silos
* A simple way to start `SENTRY_USE_SILOS=1 sentry devserver`
* Webpack proxying for both control and region silos.
* These changes also contain the changes from #49922

### Getting started

Before one can use siloed development server, you'll first need to
generate the split databases from your current monolith database using
`make create-db` and `bin/split-silo-database`. Then you'll be able to
start the servers.

### What's not working

* Sentry apps & integrations still contain many silo containment errors.
* Organization member operations still contain failures as we've not
finished fully separating the Organizationmember : User relationship.

### What will be fixed next

* Partitioning celery beat jobs during startup.
* Fixing up the various silo assignment issues.
* Expanding the control silo proxy paths.

bin/split-silo-database

@@ -63,20 +63,20 @@ def main(database: str, reset: bool, verbose: bool):
 
     This operation will not modify the original source database.
     """
-    region_models = []
-    control_models = []
+    region_tables = ["django_migrations"]
+    control_tables = ["django_migrations"]
     for model in apps.get_models():
         silo_limit = getattr(model._meta, "silo_limit", None)
         if not silo_limit:
             click.echo(f"> Could not find silo assignment for {model._meta.db_table}")
             continue
         if SiloMode.CONTROL in silo_limit.modes:
-            control_models.append(model._meta.db_table)
+            control_tables.append(model._meta.db_table)
         if SiloMode.REGION in silo_limit.modes:
-            region_models.append(model._meta.db_table)
+            region_tables.append(model._meta.db_table)
 
-    split_database(control_models, database, "control", reset=reset, verbose=verbose)
-    split_database(region_models, database, "region", reset=reset, verbose=verbose)
+    split_database(control_tables, database, "control", reset=reset, verbose=verbose)
+    split_database(region_tables, database, "region", reset=reset, verbose=verbose)
 
 
 if __name__ == "__main__":

scripts/lib.sh

@@ -170,6 +170,9 @@ run-dependent-services() {
 create-db() {
     echo "--> Creating 'sentry' database"
     docker exec sentry_postgres createdb -h 127.0.0.1 -U postgres -E utf-8 sentry || true
+    echo "--> Creating 'control' and 'region' database"
+    docker exec sentry_postgres createdb -h 127.0.0.1 -U postgres -E utf-8 control || true
+    docker exec sentry_postgres createdb -h 127.0.0.1 -U postgres -E utf-8 region || true
 }
 
 apply-migrations() {
@@ -221,6 +224,9 @@ clean() {
 drop-db() {
     echo "--> Dropping existing 'sentry' database"
     docker exec sentry_postgres dropdb --if-exists -h 127.0.0.1 -U postgres sentry
+    echo "--> Dropping 'control' and 'region' database"
+    docker exec sentry_postgres dropdb --if-exists -h 127.0.0.1 -U postgres control
+    docker exec sentry_postgres dropdb --if-exists -h 127.0.0.1 -U postgres region
 }
 
 reset-db() {

src/sentry/auth/superuser.py

@@ -131,6 +131,9 @@ class Superuser:
         org = getattr(self.request, "organization", None)
         if org and org.id != self.org_id:
             return self._check_expired_on_org_change()
+        # We have a wsgi request with no user.
+        if not hasattr(self.request, "user"):
+            return False
         # if we've been logged out
         if not self.request.user.is_authenticated:
             return False

src/sentry/conf/server.py

@@ -14,6 +14,7 @@ from urllib.parse import urlparse
 
 import sentry
 from sentry.types.region import Region
+from sentry.utils import json
 from sentry.utils.celery import crontab_with_minute_jitter
 from sentry.utils.types import type_from_value
 
@@ -201,6 +202,7 @@ if "DATABASE_URL" in os.environ:
     if url.scheme == "postgres":
         DATABASES["default"]["ENGINE"] = "sentry.db.postgres"
 
+
 # This should always be UTC.
 TIME_ZONE = "UTC"
 
@@ -3330,6 +3332,40 @@ SENTRY_ISSUE_PLATFORM_FUTURES_MAX_LIMIT = 10000
 SENTRY_REGION = os.environ.get("SENTRY_REGION", None)
 SENTRY_REGION_CONFIG: Union[Iterable[Region], str] = ()
 
+# Enable siloed development environment.
+USE_SILOS = os.environ.get("SENTRY_USE_SILOS", None)
+
+if USE_SILOS:
+    # Add connections for the region & control silo databases.
+    DATABASES["control"] = DATABASES["default"].copy()
+    DATABASES["control"]["NAME"] = "control"
+
+    DATABASES["region"] = DATABASES["default"].copy()
+    DATABASES["region"]["NAME"] = "region"
+
+    # Addresses are hardcoded based on the defaults
+    # we use in commands/devserver.
+    SENTRY_REGION_CONFIG = json.dumps(
+        [
+            {
+                "name": "us",
+                "snowflake_id": 1,
+                "category": "MULTI_TENANT",
+                "address": "http://localhost:8000",
+                "api_token": "dev-region-silo-token",
+            }
+        ]
+    )
+    control_port = os.environ.get("SENTRY_CONTROL_SILO_PORT", "8010")
+    DEV_HYBRID_CLOUD_RPC_SENDER = json.dumps(
+        {
+            "is_allowed": True,
+            "control_silo_api_token": "dev-control-silo-token",
+            "control_silo_address": f"http://127.0.0.1:{control_port}",
+        }
+    )
+    DATABASE_ROUTERS = ("sentry.db.router.SiloRouter",)
+
 # How long we should wait for a gateway proxy request to return before giving up
 GATEWAY_PROXY_TIMEOUT = None
 

src/sentry/db/router.py

@@ -1,3 +1,4 @@
+import logging
 import sys
 from typing import List
 
@@ -8,6 +9,8 @@ from django.db.utils import ConnectionDoesNotExist
 from sentry.db.models.base import Model
 from sentry.silo.base import SiloMode
 
+logger = logging.getLogger(__name__)
+
 
 class SiloRouter:
     """
@@ -43,10 +46,12 @@ class SiloRouter:
         try:
             # By accessing the connections Django will raise
             # Use `assert` to appease linters
-            assert connections["control"]
             assert connections["region"]
+            assert connections["control"]
             self.__is_simulated = True
-        except (AssertionError, ConnectionDoesNotExist):
+            logging.debug("Using simulated silos")
+        except (AssertionError, ConnectionDoesNotExist) as err:
+            logging.debug("Cannot use simulated silos", extra={"error": str(err)})
             self.__is_simulated = False
 
     def use_simulated(self, value: bool):
@@ -67,6 +72,7 @@ class SiloRouter:
                 return self.__simulated_map[silo_mode]
             if active_mode == silo_mode:
                 return "default"
+
             raise ValueError(
                 f"Cannot resolve table {table} in {silo_mode}. "
                 f"Application silo mode is {active_mode} and simulated silos are not enabled."

src/sentry/runner/commands/devserver.py

@@ -201,6 +201,8 @@ and run `sentry devservices up kafka zookeeper`.
     needs_https = parsed_url.scheme == "https" and (parsed_url.port or 443) > 1024
     has_https = shutil.which("https") is not None
 
+    control_silo_port = port + 10
+
     if needs_https and not has_https:
         from sentry.runner.initializer import show_big_error
 
@@ -249,6 +251,7 @@ and run `sentry devservices up kafka zookeeper`.
 
         proxy_port = port
         port = port + 1
+        control_silo_port = control_silo_port + 1
 
         uwsgi_overrides["protocol"] = "http"
 
@@ -256,6 +259,8 @@ and run `sentry devservices up kafka zookeeper`.
         os.environ["SENTRY_WEBPACK_PROXY_HOST"] = "%s" % host
         os.environ["SENTRY_WEBPACK_PROXY_PORT"] = "%s" % proxy_port
         os.environ["SENTRY_BACKEND_PORT"] = "%s" % port
+        if settings.USE_SILOS:
+            os.environ["SENTRY_CONTROL_SILO_PORT"] = str(control_silo_port)
 
         # webpack and/or typescript is causing memory issues
         os.environ["NODE_OPTIONS"] = (
@@ -277,6 +282,10 @@ and run `sentry devservices up kafka zookeeper`.
 
     os.environ["SENTRY_USE_RELAY"] = "1" if settings.SENTRY_USE_RELAY else ""
 
+    if settings.USE_SILOS:
+        os.environ["SENTRY_SILO_MODE"] = "REGION"
+        os.environ["SENTRY_REGION"] = "us"
+
     if workers:
         if settings.CELERY_ALWAYS_EAGER:
             raise click.ClickException(
@@ -372,7 +381,7 @@ and run `sentry devservices up kafka zookeeper`.
 
     # If we don't need any other daemons, just launch a normal uwsgi webserver
     # and avoid dealing with subprocesses
-    if not daemons:
+    if not daemons and not settings.USE_SILOS:
         server.run()
 
     import sys
@@ -409,5 +418,32 @@ and run `sentry devservices up kafka zookeeper`.
         )
         manager.add_process(name, list2cmdline(cmd), quiet=quiet, cwd=cwd)
 
+    if settings.USE_SILOS:
+        control_environ = {
+            "SENTRY_SILO_MODE": "CONTROL",
+            "SENTRY_REGION": "",
+            "SENTRY_DEVSERVER_BIND": f"localhost:{control_silo_port}",
+            # Override variable set by SentryHTTPServer.prepare_environment()
+            "UWSGI_HTTP_SOCKET": f"127.0.0.1:{control_silo_port}",
+        }
+        merged_env = os.environ.copy()
+        merged_env.update(control_environ)
+        control_services = ["server"]
+        if workers:
+            # TODO(hybridcloud) The cron processes don't work in siloed mode yet.
+            # Both silos will spawn crons for the other silo. We need to filter
+            # the cron job list during application configuration
+            control_services.extend(["cron", "worker"])
+
+        for service in control_services:
+            name, cmd = _get_daemon(service)
+            name = f"control.{name}"
+            quiet = (
+                name not in settings.DEVSERVER_LOGS_ALLOWLIST
+                if settings.DEVSERVER_LOGS_ALLOWLIST is not None
+                else False
+            )
+            manager.add_process(name, list2cmdline(cmd), quiet=quiet, cwd=cwd, env=merged_env)
+
     manager.loop()
     sys.exit(manager.returncode)

src/sentry/runner/initializer.py

@@ -370,9 +370,7 @@ def initialize_app(config: dict[str, Any], skip_service_validation: bool = False
 
     django.setup()
 
-    if getattr(settings, "SENTRY_REGION_CONFIG", None) is not None:
-        for region in settings.SENTRY_REGION_CONFIG:
-            region.validate()
+    validate_regions(settings)
 
     monkeypatch_django_migrations()
 
@@ -454,6 +452,27 @@ def validate_options(settings: Any) -> None:
     default_manager.validate(settings.SENTRY_OPTIONS, warn=True)
 
 
+def validate_regions(settings: Any) -> None:
+    from sentry.types.region import Region, RegionCategory
+    from sentry.utils import json
+
+    region_config = getattr(settings, "SENTRY_REGION_CONFIG", None)
+    if not region_config:
+        return
+
+    if isinstance(region_config, str):
+        parsed = []
+        config_values = json.loads(region_config)
+        for config_value in config_values:
+            config_value["category"] = RegionCategory[config_value["category"]]
+            parsed.append(Region(**config_value))
+
+        settings.SENTRY_REGION_CONFIG = parsed
+    else:
+        for region in region_config:
+            region.validate()
+
+
 import django.db.models.base
 
 model_unpickle = django.db.models.base.model_unpickle

src/sentry/services/hybrid_cloud/integration/service.py

@@ -5,7 +5,7 @@
 
 from abc import abstractmethod
 from datetime import datetime
-from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, Union, cast
+from typing import Any, Dict, List, Mapping, Optional, Tuple, Union, cast
 
 from sentry.integrations.base import (
     IntegrationFeatures,
@@ -71,7 +71,7 @@ class IntegrationService(RpcService):
     def get_integrations(
         self,
         *,
-        integration_ids: Optional[Iterable[int]] = None,
+        integration_ids: Optional[List[int]] = None,
         organization_id: Optional[int] = None,
         status: Optional[int] = None,
         providers: Optional[List[str]] = None,

src/sentry/services/hybrid_cloud/rpc.py

@@ -427,11 +427,11 @@ def dispatch_to_local_service(
         if isinstance(value, RpcModel):
             return value.dict()
 
+        if isinstance(value, dict):
+            return {key: result_to_dict(val) for key, val in value.items()}
+
         if isinstance(value, Iterable) and not isinstance(value, str):
-            output = []
-            for item in value:
-                output.append(result_to_dict(item))
-            return output
+            return [result_to_dict(item) for item in value]
 
         return value
 
@@ -486,7 +486,8 @@ def _fire_request(url: str, body: Any, api_token: str) -> urllib.response.addinf
     request = Request(url)
     request.add_header("Content-Type", f"application/json; charset={_RPC_CONTENT_CHARSET}")
     request.add_header("Content-Length", str(len(data)))
-    request.add_header("Authorization", f"Bearer {api_token}")
+    # TODO(hybridcloud) Re-enable this when we've implemented RPC authentication
+    # request.add_header("Authorization", f"Bearer {api_token}")
     return urlopen(request, data)  # type: ignore
 
 

webpack.config.ts

@@ -56,6 +56,7 @@ const IS_DEPLOY_PREVIEW = !!env.NOW_GITHUB_DEPLOYMENT;
 const IS_UI_DEV_ONLY = !!env.SENTRY_UI_DEV_ONLY;
 const DEV_MODE = !(IS_PRODUCTION || IS_CI);
 const WEBPACK_MODE: Configuration['mode'] = IS_PRODUCTION ? 'production' : 'development';
+const CONTROL_SILO_PORT = env.SENTRY_CONTROL_SILO_PORT;
 
 // Environment variables that are used by other tooling and should
 // not be user configurable.
@@ -555,6 +556,31 @@ if (
     const backendAddress = `http://127.0.0.1:${SENTRY_BACKEND_PORT}/`;
     const relayAddress = 'http://127.0.0.1:7899';
 
+    // If we're running siloed servers we also need to proxy
+    // those requests to the right server.
+    let controlSiloProxy = {};
+    if (CONTROL_SILO_PORT) {
+      // TODO(hybridcloud) We also need to use this URL pattern
+      // list to select contro/region when making API requests in non-proxied
+      // environments (like production). We'll likely need a way to consolidate this
+      // with the configuration api.Client uses.
+      const controlSiloAddress = `http://127.0.0.1:${CONTROL_SILO_PORT}`;
+      controlSiloProxy = {
+        '/api/0/users/**': controlSiloAddress,
+        '/api/0/sentry-apps/**': controlSiloAddress,
+        '/api/0/organizations/*/audit-logs/**': controlSiloAddress,
+        '/api/0/organizations/*/broadcasts/**': controlSiloAddress,
+        '/api/0/organizations/*/integrations/**': controlSiloAddress,
+        '/api/0/organizations/*/config/integrations/**': controlSiloAddress,
+        '/api/0/organizations/*/sentry-apps/**': controlSiloAddress,
+        '/api/0/organizations/*/sentry-app-installations/**': controlSiloAddress,
+        '/api/0/api-authorizations/**': controlSiloAddress,
+        '/api/0/api-applications/**': controlSiloAddress,
+        '/api/0/doc-integrations/**': controlSiloAddress,
+        '/api/0/assistant/**': controlSiloAddress,
+      };
+    }
+
     appConfig.devServer = {
       ...appConfig.devServer,
       static: {
@@ -563,6 +589,7 @@ if (
       },
       // syntax for matching is using https://www.npmjs.com/package/micromatch
       proxy: {
+        ...controlSiloProxy,
         '/api/store/**': relayAddress,
         '/api/{1..9}*({0..9})/**': relayAddress,
         '/api/0/relays/outcomes/': relayAddress,