Coverage for app/backend/src/app.py: 0%
106 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import logging
2import signal
3import sys
4import threading
5from multiprocessing import Process
6from os import environ
7from tempfile import TemporaryDirectory
8from types import TracebackType
10# these two lines need to be at the top of the file before we span child processes
11# this temp dir will be destroyed when prometheus_multiproc_dir is destroyed, aka at the end of the program.
12# Also note that this should only be done in the main process.
13if __name__ == "__main__":
14 prometheus_multiproc_dir = TemporaryDirectory()
15 environ["PROMETHEUS_MULTIPROC_DIR"] = prometheus_multiproc_dir.name
17# ruff: noqa: E402
19import sentry_sdk
20from sentry_sdk.integrations import excepthook
21from sentry_sdk.integrations.logging import LoggingIntegration
22from sqlalchemy.sql import text
24from couchers.config import config
25from couchers.constants import API_BASE_PORT, API_WORKER_COUNT, GRACEFUL_SHUTDOWN_TIMEOUT, MEDIA_PORT
26from couchers.db import apply_migrations, db_post_fork, session_scope
27from couchers.experimentation import setup_experimentation
28from couchers.i18n.locales import get_main_i18next
29from couchers.jobs.worker import start_jobs_scheduler, start_jobs_worker
30from couchers.metrics import create_prometheus_server
31from couchers.profiling import setup_profiling
32from couchers.server import create_main_server, create_media_server
33from couchers.supervisor import supervise
34from couchers.tracing import setup_tracing
35from dummy_data import add_dummy_data
37config.check()
39logging.basicConfig(
40 format="[%(process)5d:%(threadName)-20s] %(asctime)s: %(name)s:%(lineno)d: %(message)s", level=logging.INFO
41)
42logger = logging.getLogger(__name__)
45def _run_api_server(port: int) -> None:
46 try:
47 db_post_fork()
48 setup_experimentation()
49 setup_tracing()
50 setup_profiling(role="api", instance=f"api-{port}")
52 server = create_main_server(port=port, start_resource_sampler=True)
53 server.start()
54 logger.info(f"API worker serving on {port}")
56 terminate = threading.Event()
57 signal.signal(signal.SIGTERM, lambda *_: terminate.set())
58 signal.signal(signal.SIGINT, lambda *_: terminate.set())
59 terminate.wait()
61 logger.info(f"API worker on {port} draining (up to {GRACEFUL_SHUTDOWN_TIMEOUT}s)")
62 server.stop(GRACEFUL_SHUTDOWN_TIMEOUT).wait()
63 except Exception:
64 # multiprocessing would only print this to stderr; send the traceback to Sentry (and flush, since
65 # the process is about to die and the parent will restart the container) before re-raising
66 sentry_sdk.capture_exception()
67 sentry_sdk.flush()
68 raise
71def start_api_worker(port: int) -> Process:
72 worker = Process(target=_run_api_server, args=(port,))
73 worker.start()
74 return worker
77def log_unhandled_exception(
78 exc_type: type[BaseException],
79 exc_value: BaseException,
80 exc_traceback: TracebackType | None,
81) -> None:
82 """Make sure that any unhandled exceptions will write to the logs"""
83 if issubclass(exc_type, KeyboardInterrupt):
84 # call the default excepthook saved at __excepthook__
85 sys.__excepthook__(exc_type, exc_value, exc_traceback)
86 return
87 logger.critical("Unhandled exception", exc_info=(exc_type, exc_value, exc_traceback))
90def common_init() -> None:
91 sys.excepthook = log_unhandled_exception
93 if config.SENTRY_ENABLED:
94 # Sends exception tracebacks to Sentry, a cloud service for collecting exceptions
95 sentry_sdk.init(
96 config.SENTRY_URL,
97 traces_sample_rate=0.0,
98 environment=config.COOKIE_DOMAIN,
99 release=config.VERSION,
100 # The global excepthook picks up already handled gRPC errors (e.g. grpc.StatusCode.NOT_FOUND)
101 disabled_integrations=[
102 excepthook.ExcepthookIntegration(),
103 ],
104 integrations=[
105 LoggingIntegration(event_level=logging.CRITICAL),
106 ],
107 )
109 logger.info("Checking DB connection")
110 with session_scope() as session:
111 res = session.execute(text("SELECT 42;"))
112 if list(res) != [(42,)]:
113 raise Exception("Failed to connect to DB")
116def main() -> None:
117 logger.info("Running DB migrations")
119 apply_migrations()
121 get_main_i18next() # Force eager loading of translations
123 if config.ADD_DUMMY_DATA:
124 add_dummy_data()
126 logger.info("Starting")
128 children: list[Process] = []
130 if config.ROLE in ["scheduler", "all"]:
131 scheduler = start_jobs_scheduler()
132 scheduler.name = "scheduler"
133 children.append(scheduler)
135 if config.ROLE in ["worker", "all"]:
136 for i in range(config.BACKGROUND_WORKER_PROCESSES):
137 worker = start_jobs_worker(i, config.BACKGROUND_WORKER_THREADS_PER_PROCESS)
138 worker.name = f"worker-{i}"
139 children.append(worker)
141 # The multiprocessing start method is forkserver/spawn (Python 3.14 default; never
142 # fork), so each worker runs its own per-process init — don't pin set_start_method("fork") to "simplify"
143 # this, that reintroduces fork-after-threads hazards.
144 if config.ROLE in ["api", "all"]:
145 for port in range(API_BASE_PORT, API_BASE_PORT + API_WORKER_COUNT):
146 api_worker = start_api_worker(port)
147 api_worker.name = f"api-{port}"
148 children.append(api_worker)
150 create_prometheus_server(8000)
152 # Must precede setup_tracing(), which reads the `trace_sample_ratio` flag.
153 setup_experimentation()
155 setup_tracing()
157 media_server = None
158 if config.ROLE in ["api", "all"]:
159 media_server = create_media_server(port=MEDIA_PORT)
160 media_server.start()
161 logger.info(f"Media server serving on {MEDIA_PORT}")
163 logger.info("App started, supervising child processes")
164 crashed = supervise(children, parent_servers=[media_server] if media_server is not None else [])
166 if crashed is not None:
167 sys.exit(1)
170if __name__ == "__main__":
171 common_init()
172 main()
173elif __name__ == "__mp_main__": # processes created via multiprocessing
174 common_init()