1
0

configure_workers_and_start.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717
  1. #!/usr/bin/env python
  2. # Copyright 2021 The Matrix.org Foundation C.I.C.
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. # This script reads environment variables and generates a shared Synapse worker,
  16. # nginx and supervisord configs depending on the workers requested.
  17. #
  18. # The environment variables it reads are:
  19. # * SYNAPSE_SERVER_NAME: The desired server_name of the homeserver.
  20. # * SYNAPSE_REPORT_STATS: Whether to report stats.
  21. # * SYNAPSE_WORKER_TYPES: A comma separated list of worker names as specified in WORKER_CONFIG
  22. # below. Leave empty for no workers.
  23. # * SYNAPSE_AS_REGISTRATION_DIR: If specified, a directory in which .yaml and .yml files
  24. # will be treated as Application Service registration files.
  25. # * SYNAPSE_TLS_CERT: Path to a TLS certificate in PEM format.
  26. # * SYNAPSE_TLS_KEY: Path to a TLS key. If this and SYNAPSE_TLS_CERT are specified,
  27. # Nginx will be configured to serve TLS on port 8448.
  28. # * SYNAPSE_USE_EXPERIMENTAL_FORKING_LAUNCHER: Whether to use the forking launcher,
  29. # only intended for usage in Complement at the moment.
  30. # No stability guarantees are provided.
  31. # * SYNAPSE_LOG_LEVEL: Set this to DEBUG, INFO, WARNING or ERROR to change the
  32. # log level. INFO is the default.
  33. # * SYNAPSE_LOG_SENSITIVE: If unset, SQL and SQL values won't be logged,
  34. # regardless of the SYNAPSE_LOG_LEVEL setting.
  35. #
  36. # NOTE: According to Complement's ENTRYPOINT expectations for a homeserver image (as defined
  37. # in the project's README), this script may be run multiple times, and functionality should
  38. # continue to work if so.
  39. import os
  40. import platform
  41. import subprocess
  42. import sys
  43. from pathlib import Path
  44. from typing import Any, Dict, List, Mapping, MutableMapping, NoReturn, Optional, Set
  45. import yaml
  46. from jinja2 import Environment, FileSystemLoader
  47. MAIN_PROCESS_HTTP_LISTENER_PORT = 8080
  48. # Workers with exposed endpoints needs either "client", "federation", or "media" listener_resources
  49. # Watching /_matrix/client needs a "client" listener
  50. # Watching /_matrix/federation needs a "federation" listener
  51. # Watching /_matrix/media and related needs a "media" listener
  52. # Stream Writers require "client" and "replication" listeners because they
  53. # have to attach by instance_map to the master process and have client endpoints.
  54. WORKERS_CONFIG: Dict[str, Dict[str, Any]] = {
  55. "pusher": {
  56. "app": "synapse.app.generic_worker",
  57. "listener_resources": [],
  58. "endpoint_patterns": [],
  59. "shared_extra_conf": {},
  60. "worker_extra_conf": "",
  61. },
  62. "user_dir": {
  63. "app": "synapse.app.generic_worker",
  64. "listener_resources": ["client"],
  65. "endpoint_patterns": [
  66. "^/_matrix/client/(api/v1|r0|v3|unstable)/user_directory/search$"
  67. ],
  68. "shared_extra_conf": {"update_user_directory_from_worker": "user_dir1"},
  69. "worker_extra_conf": "",
  70. },
  71. "media_repository": {
  72. "app": "synapse.app.media_repository",
  73. "listener_resources": ["media"],
  74. "endpoint_patterns": [
  75. "^/_matrix/media/",
  76. "^/_synapse/admin/v1/purge_media_cache$",
  77. "^/_synapse/admin/v1/room/.*/media.*$",
  78. "^/_synapse/admin/v1/user/.*/media.*$",
  79. "^/_synapse/admin/v1/media/.*$",
  80. "^/_synapse/admin/v1/quarantine_media/.*$",
  81. ],
  82. # The first configured media worker will run the media background jobs
  83. "shared_extra_conf": {
  84. "enable_media_repo": False,
  85. "media_instance_running_background_jobs": "media_repository1",
  86. },
  87. "worker_extra_conf": "enable_media_repo: true",
  88. },
  89. "appservice": {
  90. "app": "synapse.app.generic_worker",
  91. "listener_resources": [],
  92. "endpoint_patterns": [],
  93. "shared_extra_conf": {"notify_appservices_from_worker": "appservice1"},
  94. "worker_extra_conf": "",
  95. },
  96. "federation_sender": {
  97. "app": "synapse.app.generic_worker",
  98. "listener_resources": [],
  99. "endpoint_patterns": [],
  100. "shared_extra_conf": {},
  101. "worker_extra_conf": "",
  102. },
  103. "synchrotron": {
  104. "app": "synapse.app.generic_worker",
  105. "listener_resources": ["client"],
  106. "endpoint_patterns": [
  107. "^/_matrix/client/(v2_alpha|r0|v3)/sync$",
  108. "^/_matrix/client/(api/v1|v2_alpha|r0|v3)/events$",
  109. "^/_matrix/client/(api/v1|r0|v3)/initialSync$",
  110. "^/_matrix/client/(api/v1|r0|v3)/rooms/[^/]+/initialSync$",
  111. ],
  112. "shared_extra_conf": {},
  113. "worker_extra_conf": "",
  114. },
  115. "client_reader": {
  116. "app": "synapse.app.generic_worker",
  117. "listener_resources": ["client"],
  118. "endpoint_patterns": [
  119. "^/_matrix/client/(api/v1|r0|v3|unstable)/publicRooms$",
  120. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/joined_members$",
  121. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/context/.*$",
  122. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/members$",
  123. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/state$",
  124. "^/_matrix/client/v1/rooms/.*/hierarchy$",
  125. "^/_matrix/client/(v1|unstable)/rooms/.*/relations/",
  126. "^/_matrix/client/v1/rooms/.*/threads$",
  127. "^/_matrix/client/(api/v1|r0|v3|unstable)/login$",
  128. "^/_matrix/client/(api/v1|r0|v3|unstable)/account/3pid$",
  129. "^/_matrix/client/(api/v1|r0|v3|unstable)/account/whoami$",
  130. "^/_matrix/client/versions$",
  131. "^/_matrix/client/(api/v1|r0|v3|unstable)/voip/turnServer$",
  132. "^/_matrix/client/(r0|v3|unstable)/register$",
  133. "^/_matrix/client/(r0|v3|unstable)/auth/.*/fallback/web$",
  134. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/messages$",
  135. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/event",
  136. "^/_matrix/client/(api/v1|r0|v3|unstable)/joined_rooms",
  137. "^/_matrix/client/(api/v1|r0|v3|unstable/.*)/rooms/.*/aliases",
  138. "^/_matrix/client/v1/rooms/.*/timestamp_to_event$",
  139. "^/_matrix/client/(api/v1|r0|v3|unstable)/search",
  140. "^/_matrix/client/(r0|v3|unstable)/user/.*/filter(/|$)",
  141. ],
  142. "shared_extra_conf": {},
  143. "worker_extra_conf": "",
  144. },
  145. "federation_reader": {
  146. "app": "synapse.app.generic_worker",
  147. "listener_resources": ["federation"],
  148. "endpoint_patterns": [
  149. "^/_matrix/federation/(v1|v2)/event/",
  150. "^/_matrix/federation/(v1|v2)/state/",
  151. "^/_matrix/federation/(v1|v2)/state_ids/",
  152. "^/_matrix/federation/(v1|v2)/backfill/",
  153. "^/_matrix/federation/(v1|v2)/get_missing_events/",
  154. "^/_matrix/federation/(v1|v2)/publicRooms",
  155. "^/_matrix/federation/(v1|v2)/query/",
  156. "^/_matrix/federation/(v1|v2)/make_join/",
  157. "^/_matrix/federation/(v1|v2)/make_leave/",
  158. "^/_matrix/federation/(v1|v2)/send_join/",
  159. "^/_matrix/federation/(v1|v2)/send_leave/",
  160. "^/_matrix/federation/(v1|v2)/invite/",
  161. "^/_matrix/federation/(v1|v2)/query_auth/",
  162. "^/_matrix/federation/(v1|v2)/event_auth/",
  163. "^/_matrix/federation/v1/timestamp_to_event/",
  164. "^/_matrix/federation/(v1|v2)/exchange_third_party_invite/",
  165. "^/_matrix/federation/(v1|v2)/user/devices/",
  166. "^/_matrix/federation/(v1|v2)/get_groups_publicised$",
  167. "^/_matrix/key/v2/query",
  168. ],
  169. "shared_extra_conf": {},
  170. "worker_extra_conf": "",
  171. },
  172. "federation_inbound": {
  173. "app": "synapse.app.generic_worker",
  174. "listener_resources": ["federation"],
  175. "endpoint_patterns": ["/_matrix/federation/(v1|v2)/send/"],
  176. "shared_extra_conf": {},
  177. "worker_extra_conf": "",
  178. },
  179. "event_persister": {
  180. "app": "synapse.app.generic_worker",
  181. "listener_resources": ["replication"],
  182. "endpoint_patterns": [],
  183. "shared_extra_conf": {},
  184. "worker_extra_conf": "",
  185. },
  186. "background_worker": {
  187. "app": "synapse.app.generic_worker",
  188. "listener_resources": [],
  189. "endpoint_patterns": [],
  190. # This worker cannot be sharded. Therefore there should only ever be one background
  191. # worker, and it should be named background_worker1
  192. "shared_extra_conf": {"run_background_tasks_on": "background_worker1"},
  193. "worker_extra_conf": "",
  194. },
  195. "event_creator": {
  196. "app": "synapse.app.generic_worker",
  197. "listener_resources": ["client"],
  198. "endpoint_patterns": [
  199. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/redact",
  200. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/send",
  201. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/(join|invite|leave|ban|unban|kick)$",
  202. "^/_matrix/client/(api/v1|r0|v3|unstable)/join/",
  203. "^/_matrix/client/(api/v1|r0|v3|unstable)/knock/",
  204. "^/_matrix/client/(api/v1|r0|v3|unstable)/profile/",
  205. "^/_matrix/client/(v1|unstable/org.matrix.msc2716)/rooms/.*/batch_send",
  206. ],
  207. "shared_extra_conf": {},
  208. "worker_extra_conf": "",
  209. },
  210. "frontend_proxy": {
  211. "app": "synapse.app.generic_worker",
  212. "listener_resources": ["client", "replication"],
  213. "endpoint_patterns": ["^/_matrix/client/(api/v1|r0|v3|unstable)/keys/upload"],
  214. "shared_extra_conf": {},
  215. "worker_extra_conf": "",
  216. },
  217. "account_data": {
  218. "app": "synapse.app.generic_worker",
  219. "listener_resources": ["client", "replication"],
  220. "endpoint_patterns": [
  221. "^/_matrix/client/(r0|v3|unstable)/.*/tags",
  222. "^/_matrix/client/(r0|v3|unstable)/.*/account_data",
  223. ],
  224. "shared_extra_conf": {},
  225. "worker_extra_conf": "",
  226. },
  227. "presence": {
  228. "app": "synapse.app.generic_worker",
  229. "listener_resources": ["client", "replication"],
  230. "endpoint_patterns": ["^/_matrix/client/(api/v1|r0|v3|unstable)/presence/"],
  231. "shared_extra_conf": {},
  232. "worker_extra_conf": "",
  233. },
  234. "receipts": {
  235. "app": "synapse.app.generic_worker",
  236. "listener_resources": ["client", "replication"],
  237. "endpoint_patterns": [
  238. "^/_matrix/client/(r0|v3|unstable)/rooms/.*/receipt",
  239. "^/_matrix/client/(r0|v3|unstable)/rooms/.*/read_markers",
  240. ],
  241. "shared_extra_conf": {},
  242. "worker_extra_conf": "",
  243. },
  244. "to_device": {
  245. "app": "synapse.app.generic_worker",
  246. "listener_resources": ["client", "replication"],
  247. "endpoint_patterns": ["^/_matrix/client/(r0|v3|unstable)/sendToDevice/"],
  248. "shared_extra_conf": {},
  249. "worker_extra_conf": "",
  250. },
  251. "typing": {
  252. "app": "synapse.app.generic_worker",
  253. "listener_resources": ["client", "replication"],
  254. "endpoint_patterns": [
  255. "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/typing"
  256. ],
  257. "shared_extra_conf": {},
  258. "worker_extra_conf": "",
  259. },
  260. }
  261. # Templates for sections that may be inserted multiple times in config files
  262. NGINX_LOCATION_CONFIG_BLOCK = """
  263. location ~* {endpoint} {{
  264. proxy_pass {upstream};
  265. proxy_set_header X-Forwarded-For $remote_addr;
  266. proxy_set_header X-Forwarded-Proto $scheme;
  267. proxy_set_header Host $host;
  268. }}
  269. """
  270. NGINX_UPSTREAM_CONFIG_BLOCK = """
  271. upstream {upstream_worker_type} {{
  272. {body}
  273. }}
  274. """
  275. # Utility functions
  276. def log(txt: str) -> None:
  277. print(txt)
  278. def error(txt: str) -> NoReturn:
  279. print(txt, file=sys.stderr)
  280. sys.exit(2)
  281. def flush_buffers() -> None:
  282. sys.stdout.flush()
  283. sys.stderr.flush()
  284. def convert(src: str, dst: str, **template_vars: object) -> None:
  285. """Generate a file from a template
  286. Args:
  287. src: Path to the input file.
  288. dst: Path to write to.
  289. template_vars: The arguments to replace placeholder variables in the template with.
  290. """
  291. # Read the template file
  292. # We disable autoescape to prevent template variables from being escaped,
  293. # as we're not using HTML.
  294. env = Environment(loader=FileSystemLoader(os.path.dirname(src)), autoescape=False)
  295. template = env.get_template(os.path.basename(src))
  296. # Generate a string from the template.
  297. rendered = template.render(**template_vars)
  298. # Write the generated contents to a file
  299. #
  300. # We use append mode in case the files have already been written to by something else
  301. # (for instance, as part of the instructions in a dockerfile).
  302. with open(dst, "a") as outfile:
  303. # In case the existing file doesn't end with a newline
  304. outfile.write("\n")
  305. outfile.write(rendered)
  306. def add_worker_roles_to_shared_config(
  307. shared_config: dict,
  308. worker_type: str,
  309. worker_name: str,
  310. worker_port: int,
  311. ) -> None:
  312. """Given a dictionary representing a config file shared across all workers,
  313. append appropriate worker information to it for the current worker_type instance.
  314. Args:
  315. shared_config: The config dict that all worker instances share (after being converted to YAML)
  316. worker_type: The type of worker (one of those defined in WORKERS_CONFIG).
  317. worker_name: The name of the worker instance.
  318. worker_port: The HTTP replication port that the worker instance is listening on.
  319. """
  320. # The instance_map config field marks the workers that write to various replication streams
  321. instance_map = shared_config.setdefault("instance_map", {})
  322. # Worker-type specific sharding config
  323. if worker_type == "pusher":
  324. shared_config.setdefault("pusher_instances", []).append(worker_name)
  325. elif worker_type == "federation_sender":
  326. shared_config.setdefault("federation_sender_instances", []).append(worker_name)
  327. elif worker_type == "event_persister":
  328. # Event persisters write to the events stream, so we need to update
  329. # the list of event stream writers
  330. shared_config.setdefault("stream_writers", {}).setdefault("events", []).append(
  331. worker_name
  332. )
  333. # Map of stream writer instance names to host/ports combos
  334. instance_map[worker_name] = {
  335. "host": "localhost",
  336. "port": worker_port,
  337. }
  338. elif worker_type in ["account_data", "presence", "receipts", "to_device", "typing"]:
  339. # Update the list of stream writers
  340. # It's convenient that the name of the worker type is the same as the stream to write
  341. shared_config.setdefault("stream_writers", {}).setdefault(
  342. worker_type, []
  343. ).append(worker_name)
  344. # Map of stream writer instance names to host/ports combos
  345. # For now, all stream writers need http replication ports
  346. instance_map[worker_name] = {
  347. "host": "localhost",
  348. "port": worker_port,
  349. }
  350. def generate_base_homeserver_config() -> None:
  351. """Starts Synapse and generates a basic homeserver config, which will later be
  352. modified for worker support.
  353. Raises: CalledProcessError if calling start.py returned a non-zero exit code.
  354. """
  355. # start.py already does this for us, so just call that.
  356. # note that this script is copied in in the official, monolith dockerfile
  357. os.environ["SYNAPSE_HTTP_PORT"] = str(MAIN_PROCESS_HTTP_LISTENER_PORT)
  358. subprocess.run(["/usr/local/bin/python", "/start.py", "migrate_config"], check=True)
  359. def generate_worker_files(
  360. environ: Mapping[str, str], config_path: str, data_dir: str
  361. ) -> None:
  362. """Read the desired list of workers from environment variables and generate
  363. shared homeserver, nginx and supervisord configs.
  364. Args:
  365. environ: os.environ instance.
  366. config_path: The location of the generated Synapse main worker config file.
  367. data_dir: The location of the synapse data directory. Where log and
  368. user-facing config files live.
  369. """
  370. # Note that yaml cares about indentation, so care should be taken to insert lines
  371. # into files at the correct indentation below.
  372. # shared_config is the contents of a Synapse config file that will be shared amongst
  373. # the main Synapse process as well as all workers.
  374. # It is intended mainly for disabling functionality when certain workers are spun up,
  375. # and adding a replication listener.
  376. # First read the original config file and extract the listeners block. Then we'll add
  377. # another listener for replication. Later we'll write out the result to the shared
  378. # config file.
  379. listeners = [
  380. {
  381. "port": 9093,
  382. "bind_address": "127.0.0.1",
  383. "type": "http",
  384. "resources": [{"names": ["replication"]}],
  385. }
  386. ]
  387. with open(config_path) as file_stream:
  388. original_config = yaml.safe_load(file_stream)
  389. original_listeners = original_config.get("listeners")
  390. if original_listeners:
  391. listeners += original_listeners
  392. # The shared homeserver config. The contents of which will be inserted into the
  393. # base shared worker jinja2 template.
  394. #
  395. # This config file will be passed to all workers, included Synapse's main process.
  396. shared_config: Dict[str, Any] = {"listeners": listeners}
  397. # List of dicts that describe workers.
  398. # We pass this to the Supervisor template later to generate the appropriate
  399. # program blocks.
  400. worker_descriptors: List[Dict[str, Any]] = []
  401. # Upstreams for load-balancing purposes. This dict takes the form of a worker type to the
  402. # ports of each worker. For example:
  403. # {
  404. # worker_type: {1234, 1235, ...}}
  405. # }
  406. # and will be used to construct 'upstream' nginx directives.
  407. nginx_upstreams: Dict[str, Set[int]] = {}
  408. # A map of: {"endpoint": "upstream"}, where "upstream" is a str representing what will be
  409. # placed after the proxy_pass directive. The main benefit to representing this data as a
  410. # dict over a str is that we can easily deduplicate endpoints across multiple instances
  411. # of the same worker.
  412. #
  413. # An nginx site config that will be amended to depending on the workers that are
  414. # spun up. To be placed in /etc/nginx/conf.d.
  415. nginx_locations = {}
  416. # Read the desired worker configuration from the environment
  417. worker_types_env = environ.get("SYNAPSE_WORKER_TYPES", "").strip()
  418. if not worker_types_env:
  419. # No workers, just the main process
  420. worker_types = []
  421. else:
  422. # Split type names by comma, ignoring whitespace.
  423. worker_types = [x.strip() for x in worker_types_env.split(",")]
  424. # Create the worker configuration directory if it doesn't already exist
  425. os.makedirs("/conf/workers", exist_ok=True)
  426. # Start worker ports from this arbitrary port
  427. worker_port = 18009
  428. # A counter of worker_type -> int. Used for determining the name for a given
  429. # worker type when generating its config file, as each worker's name is just
  430. # worker_type + instance #
  431. worker_type_counter: Dict[str, int] = {}
  432. # A list of internal endpoints to healthcheck, starting with the main process
  433. # which exists even if no workers do.
  434. healthcheck_urls = ["http://localhost:8080/health"]
  435. # For each worker type specified by the user, create config values
  436. for worker_type in worker_types:
  437. worker_config = WORKERS_CONFIG.get(worker_type)
  438. if worker_config:
  439. worker_config = worker_config.copy()
  440. else:
  441. error(worker_type + " is an unknown worker type! Please fix!")
  442. new_worker_count = worker_type_counter.setdefault(worker_type, 0) + 1
  443. worker_type_counter[worker_type] = new_worker_count
  444. # Name workers by their type concatenated with an incrementing number
  445. # e.g. federation_reader1
  446. worker_name = worker_type + str(new_worker_count)
  447. worker_config.update(
  448. {"name": worker_name, "port": str(worker_port), "config_path": config_path}
  449. )
  450. # Update the shared config with any worker-type specific options
  451. shared_config.update(worker_config["shared_extra_conf"])
  452. healthcheck_urls.append("http://localhost:%d/health" % (worker_port,))
  453. # Check if more than one instance of this worker type has been specified
  454. worker_type_total_count = worker_types.count(worker_type)
  455. # Update the shared config with sharding-related options if necessary
  456. add_worker_roles_to_shared_config(
  457. shared_config, worker_type, worker_name, worker_port
  458. )
  459. # Enable the worker in supervisord
  460. worker_descriptors.append(worker_config)
  461. # Add nginx location blocks for this worker's endpoints (if any are defined)
  462. for pattern in worker_config["endpoint_patterns"]:
  463. # Determine whether we need to load-balance this worker
  464. if worker_type_total_count > 1:
  465. # Create or add to a load-balanced upstream for this worker
  466. nginx_upstreams.setdefault(worker_type, set()).add(worker_port)
  467. # Upstreams are named after the worker_type
  468. upstream = "http://" + worker_type
  469. else:
  470. upstream = "http://localhost:%d" % (worker_port,)
  471. # Note that this endpoint should proxy to this upstream
  472. nginx_locations[pattern] = upstream
  473. # Write out the worker's logging config file
  474. log_config_filepath = generate_worker_log_config(environ, worker_name, data_dir)
  475. # Then a worker config file
  476. convert(
  477. "/conf/worker.yaml.j2",
  478. "/conf/workers/{name}.yaml".format(name=worker_name),
  479. **worker_config,
  480. worker_log_config_filepath=log_config_filepath,
  481. )
  482. worker_port += 1
  483. # Build the nginx location config blocks
  484. nginx_location_config = ""
  485. for endpoint, upstream in nginx_locations.items():
  486. nginx_location_config += NGINX_LOCATION_CONFIG_BLOCK.format(
  487. endpoint=endpoint,
  488. upstream=upstream,
  489. )
  490. # Determine the load-balancing upstreams to configure
  491. nginx_upstream_config = ""
  492. for upstream_worker_type, upstream_worker_ports in nginx_upstreams.items():
  493. body = ""
  494. for port in upstream_worker_ports:
  495. body += " server localhost:%d;\n" % (port,)
  496. # Add to the list of configured upstreams
  497. nginx_upstream_config += NGINX_UPSTREAM_CONFIG_BLOCK.format(
  498. upstream_worker_type=upstream_worker_type,
  499. body=body,
  500. )
  501. # Finally, we'll write out the config files.
  502. # log config for the master process
  503. master_log_config = generate_worker_log_config(environ, "master", data_dir)
  504. shared_config["log_config"] = master_log_config
  505. # Find application service registrations
  506. appservice_registrations = None
  507. appservice_registration_dir = os.environ.get("SYNAPSE_AS_REGISTRATION_DIR")
  508. if appservice_registration_dir:
  509. # Scan for all YAML files that should be application service registrations.
  510. appservice_registrations = [
  511. str(reg_path.resolve())
  512. for reg_path in Path(appservice_registration_dir).iterdir()
  513. if reg_path.suffix.lower() in (".yaml", ".yml")
  514. ]
  515. workers_in_use = len(worker_types) > 0
  516. # Shared homeserver config
  517. convert(
  518. "/conf/shared.yaml.j2",
  519. "/conf/workers/shared.yaml",
  520. shared_worker_config=yaml.dump(shared_config),
  521. appservice_registrations=appservice_registrations,
  522. enable_redis=workers_in_use,
  523. workers_in_use=workers_in_use,
  524. )
  525. # Nginx config
  526. convert(
  527. "/conf/nginx.conf.j2",
  528. "/etc/nginx/conf.d/matrix-synapse.conf",
  529. worker_locations=nginx_location_config,
  530. upstream_directives=nginx_upstream_config,
  531. tls_cert_path=os.environ.get("SYNAPSE_TLS_CERT"),
  532. tls_key_path=os.environ.get("SYNAPSE_TLS_KEY"),
  533. )
  534. # Supervisord config
  535. os.makedirs("/etc/supervisor", exist_ok=True)
  536. convert(
  537. "/conf/supervisord.conf.j2",
  538. "/etc/supervisor/supervisord.conf",
  539. main_config_path=config_path,
  540. enable_redis=workers_in_use,
  541. )
  542. convert(
  543. "/conf/synapse.supervisord.conf.j2",
  544. "/etc/supervisor/conf.d/synapse.conf",
  545. workers=worker_descriptors,
  546. main_config_path=config_path,
  547. use_forking_launcher=environ.get("SYNAPSE_USE_EXPERIMENTAL_FORKING_LAUNCHER"),
  548. )
  549. # healthcheck config
  550. convert(
  551. "/conf/healthcheck.sh.j2",
  552. "/healthcheck.sh",
  553. healthcheck_urls=healthcheck_urls,
  554. )
  555. # Ensure the logging directory exists
  556. log_dir = data_dir + "/logs"
  557. if not os.path.exists(log_dir):
  558. os.mkdir(log_dir)
  559. def generate_worker_log_config(
  560. environ: Mapping[str, str], worker_name: str, data_dir: str
  561. ) -> str:
  562. """Generate a log.config file for the given worker.
  563. Returns: the path to the generated file
  564. """
  565. # Check whether we should write worker logs to disk, in addition to the console
  566. extra_log_template_args: Dict[str, Optional[str]] = {}
  567. if environ.get("SYNAPSE_WORKERS_WRITE_LOGS_TO_DISK"):
  568. extra_log_template_args["LOG_FILE_PATH"] = f"{data_dir}/logs/{worker_name}.log"
  569. extra_log_template_args["SYNAPSE_LOG_LEVEL"] = environ.get("SYNAPSE_LOG_LEVEL")
  570. extra_log_template_args["SYNAPSE_LOG_SENSITIVE"] = environ.get(
  571. "SYNAPSE_LOG_SENSITIVE"
  572. )
  573. # Render and write the file
  574. log_config_filepath = f"/conf/workers/{worker_name}.log.config"
  575. convert(
  576. "/conf/log.config",
  577. log_config_filepath,
  578. worker_name=worker_name,
  579. **extra_log_template_args,
  580. include_worker_name_in_log_line=environ.get(
  581. "SYNAPSE_USE_EXPERIMENTAL_FORKING_LAUNCHER"
  582. ),
  583. )
  584. return log_config_filepath
  585. def main(args: List[str], environ: MutableMapping[str, str]) -> None:
  586. config_dir = environ.get("SYNAPSE_CONFIG_DIR", "/data")
  587. config_path = environ.get("SYNAPSE_CONFIG_PATH", config_dir + "/homeserver.yaml")
  588. data_dir = environ.get("SYNAPSE_DATA_DIR", "/data")
  589. # override SYNAPSE_NO_TLS, we don't support TLS in worker mode,
  590. # this needs to be handled by a frontend proxy
  591. environ["SYNAPSE_NO_TLS"] = "yes"
  592. # Generate the base homeserver config if one does not yet exist
  593. if not os.path.exists(config_path):
  594. log("Generating base homeserver config")
  595. generate_base_homeserver_config()
  596. else:
  597. log("Base homeserver config exists—not regenerating")
  598. # This script may be run multiple times (mostly by Complement, see note at top of file).
  599. # Don't re-configure workers in this instance.
  600. mark_filepath = "/conf/workers_have_been_configured"
  601. if not os.path.exists(mark_filepath):
  602. # Always regenerate all other config files
  603. log("Generating worker config files")
  604. generate_worker_files(environ, config_path, data_dir)
  605. # Mark workers as being configured
  606. with open(mark_filepath, "w") as f:
  607. f.write("")
  608. else:
  609. log("Worker config exists—not regenerating")
  610. # Lifted right out of start.py
  611. jemallocpath = "/usr/lib/%s-linux-gnu/libjemalloc.so.2" % (platform.machine(),)
  612. if os.path.isfile(jemallocpath):
  613. environ["LD_PRELOAD"] = jemallocpath
  614. else:
  615. log("Could not find %s, will not use" % (jemallocpath,))
  616. # Start supervisord, which will start Synapse, all of the configured worker
  617. # processes, redis, nginx etc. according to the config we created above.
  618. log("Starting supervisord")
  619. flush_buffers()
  620. os.execle(
  621. "/usr/local/bin/supervisord",
  622. "supervisord",
  623. "-c",
  624. "/etc/supervisor/supervisord.conf",
  625. environ,
  626. )
  627. if __name__ == "__main__":
  628. main(sys.argv, os.environ)