Mirrors
/
synapse-old
mirror of https://github.com/matrix-org/synapse.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Convert listener_resources and endpoint_patterns to Set[str]

This commit is contained in:
Olivier Wilkinson (reivilibre) 2023-11-16 15:12:18 +00:00
parent 67d4fc8b99
commit 94a85b36f7
1 changed files with 44 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
docker/configure_workers_and_start.py
View File

@ -90,8 +90,8 @@ MAIN_PROCESS_UNIX_SOCKET_PRIVATE_PATH = "/run/main_private.sock"
# have to attach by instance_map to the master process and have client endpoints. # have to attach by instance_map to the master process and have client endpoints.
@dataclass @dataclass
class WorkerTemplate: class WorkerTemplate:
listener_resources: List[str] = field(default_factory=list) listener_resources: Set[str] = field(default_factory=set)
endpoint_patterns: List[str] = field(default_factory=list) endpoint_patterns: Set[str] = field(default_factory=set)
# (worker_name) -> {} # (worker_name) -> {}
shared_extra_conf: Callable[[str], Dict[str, Any]] = lambda _worker_name: {} shared_extra_conf: Callable[[str], Dict[str, Any]] = lambda _worker_name: {}
worker_extra_conf: str = "" worker_extra_conf: str = ""
@ -100,24 +100,24 @@ class WorkerTemplate:
WORKERS_CONFIG: Dict[str, WorkerTemplate] = { WORKERS_CONFIG: Dict[str, WorkerTemplate] = {
"pusher": WorkerTemplate(), "pusher": WorkerTemplate(),
"user_dir": WorkerTemplate( "user_dir": WorkerTemplate(
listener_resources=["client"], listener_resources={"client"},
endpoint_patterns=[ endpoint_patterns={
"^/_matrix/client/(api/v1|r0|v3|unstable)/user_directory/search$" "^/_matrix/client/(api/v1|r0|v3|unstable)/user_directory/search$"
], },
shared_extra_conf=lambda worker_name: { shared_extra_conf=lambda worker_name: {
"update_user_directory_from_worker": worker_name "update_user_directory_from_worker": worker_name
}, },
), ),
"media_repository": WorkerTemplate( "media_repository": WorkerTemplate(
listener_resources=["media"], listener_resources={"media"},
endpoint_patterns=[ endpoint_patterns={
"^/_matrix/media/", "^/_matrix/media/",
"^/_synapse/admin/v1/purge_media_cache$", "^/_synapse/admin/v1/purge_media_cache$",
"^/_synapse/admin/v1/room/.*/media.*$", "^/_synapse/admin/v1/room/.*/media.*$",
"^/_synapse/admin/v1/user/.*/media.*$", "^/_synapse/admin/v1/user/.*/media.*$",
"^/_synapse/admin/v1/media/.*$", "^/_synapse/admin/v1/media/.*$",
"^/_synapse/admin/v1/quarantine_media/.*$", "^/_synapse/admin/v1/quarantine_media/.*$",
], },
# The first configured media worker will run the media background jobs # The first configured media worker will run the media background jobs
shared_extra_conf=lambda worker_name: { shared_extra_conf=lambda worker_name: {
"enable_media_repo": False, "enable_media_repo": False,
@ -132,17 +132,17 @@ WORKERS_CONFIG: Dict[str, WorkerTemplate] = {
), ),
"federation_sender": WorkerTemplate(), "federation_sender": WorkerTemplate(),
"synchrotron": WorkerTemplate( "synchrotron": WorkerTemplate(
listener_resources=["client"], listener_resources={"client"},
endpoint_patterns=[ endpoint_patterns={
"^/_matrix/client/(v2_alpha|r0|v3)/sync$", "^/_matrix/client/(v2_alpha|r0|v3)/sync$",
"^/_matrix/client/(api/v1|v2_alpha|r0|v3)/events$", "^/_matrix/client/(api/v1|v2_alpha|r0|v3)/events$",
"^/_matrix/client/(api/v1|r0|v3)/initialSync$", "^/_matrix/client/(api/v1|r0|v3)/initialSync$",
"^/_matrix/client/(api/v1|r0|v3)/rooms/[^/]+/initialSync$", "^/_matrix/client/(api/v1|r0|v3)/rooms/[^/]+/initialSync$",
], },
), ),
"client_reader": WorkerTemplate( "client_reader": WorkerTemplate(
listener_resources=["client"], listener_resources={"client"},
endpoint_patterns=[ endpoint_patterns={
"^/_matrix/client/(api/v1|r0|v3|unstable)/publicRooms$", "^/_matrix/client/(api/v1|r0|v3|unstable)/publicRooms$",
"^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/joined_members$", "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/joined_members$",
"^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/context/.*$", "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/context/.*$",
@ -170,11 +170,11 @@ WORKERS_CONFIG: Dict[str, WorkerTemplate] = {
"^/_matrix/client/(api/v1|r0|v3|unstable)/directory/room/.*$", "^/_matrix/client/(api/v1|r0|v3|unstable)/directory/room/.*$",
"^/_matrix/client/(r0|v3|unstable)/capabilities$", "^/_matrix/client/(r0|v3|unstable)/capabilities$",
"^/_matrix/client/(r0|v3|unstable)/notifications$", "^/_matrix/client/(r0|v3|unstable)/notifications$",
], },
), ),
"federation_reader": WorkerTemplate( "federation_reader": WorkerTemplate(
listener_resources=["federation"], listener_resources={"federation"},
endpoint_patterns=[ endpoint_patterns={
"^/_matrix/federation/(v1|v2)/event/", "^/_matrix/federation/(v1|v2)/event/",
"^/_matrix/federation/(v1|v2)/state/", "^/_matrix/federation/(v1|v2)/state/",
"^/_matrix/federation/(v1|v2)/state_ids/", "^/_matrix/federation/(v1|v2)/state_ids/",
@ -194,14 +194,14 @@ WORKERS_CONFIG: Dict[str, WorkerTemplate] = {
"^/_matrix/federation/(v1|v2)/user/devices/", "^/_matrix/federation/(v1|v2)/user/devices/",
"^/_matrix/federation/(v1|v2)/get_groups_publicised$", "^/_matrix/federation/(v1|v2)/get_groups_publicised$",
"^/_matrix/key/v2/query", "^/_matrix/key/v2/query",
], },
), ),
"federation_inbound": WorkerTemplate( "federation_inbound": WorkerTemplate(
listener_resources=["federation"], listener_resources={"federation"},
endpoint_patterns=["/_matrix/federation/(v1|v2)/send/"], endpoint_patterns={"/_matrix/federation/(v1|v2)/send/"},
), ),
"event_persister": WorkerTemplate( "event_persister": WorkerTemplate(
listener_resources=["replication"], listener_resources={"replication"},
), ),
"background_worker": WorkerTemplate( "background_worker": WorkerTemplate(
# This worker cannot be sharded. Therefore, there should only ever be one # This worker cannot be sharded. Therefore, there should only ever be one
@ -209,45 +209,45 @@ WORKERS_CONFIG: Dict[str, WorkerTemplate] = {
shared_extra_conf=lambda worker_name: {"run_background_tasks_on": worker_name}, shared_extra_conf=lambda worker_name: {"run_background_tasks_on": worker_name},
), ),
"event_creator": WorkerTemplate( "event_creator": WorkerTemplate(
listener_resources=["client"], listener_resources={"client"},
endpoint_patterns=[ endpoint_patterns={
"^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/redact", "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/redact",
"^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/send", "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/send",
"^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/(join|invite|leave|ban|unban|kick)$", "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/(join|invite|leave|ban|unban|kick)$",
"^/_matrix/client/(api/v1|r0|v3|unstable)/join/", "^/_matrix/client/(api/v1|r0|v3|unstable)/join/",
"^/_matrix/client/(api/v1|r0|v3|unstable)/knock/", "^/_matrix/client/(api/v1|r0|v3|unstable)/knock/",
"^/_matrix/client/(api/v1|r0|v3|unstable)/profile/", "^/_matrix/client/(api/v1|r0|v3|unstable)/profile/",
], },
), ),
"frontend_proxy": WorkerTemplate( "frontend_proxy": WorkerTemplate(
listener_resources=["client", "replication"], listener_resources={"client", "replication"},
endpoint_patterns=["^/_matrix/client/(api/v1|r0|v3|unstable)/keys/upload"], endpoint_patterns={"^/_matrix/client/(api/v1|r0|v3|unstable)/keys/upload"},
), ),
"account_data": WorkerTemplate( "account_data": WorkerTemplate(
listener_resources=["client", "replication"], listener_resources={"client", "replication"},
endpoint_patterns=[ endpoint_patterns={
"^/_matrix/client/(r0|v3|unstable)/.*/tags", "^/_matrix/client/(r0|v3|unstable)/.*/tags",
"^/_matrix/client/(r0|v3|unstable)/.*/account_data", "^/_matrix/client/(r0|v3|unstable)/.*/account_data",
], },
), ),
"presence": WorkerTemplate( "presence": WorkerTemplate(
listener_resources=["client", "replication"], listener_resources={"client", "replication"},
endpoint_patterns=["^/_matrix/client/(api/v1|r0|v3|unstable)/presence/"], endpoint_patterns={"^/_matrix/client/(api/v1|r0|v3|unstable)/presence/"},
), ),
"receipts": WorkerTemplate( "receipts": WorkerTemplate(
listener_resources=["client", "replication"], listener_resources={"client", "replication"},
endpoint_patterns=[ endpoint_patterns={
"^/_matrix/client/(r0|v3|unstable)/rooms/.*/receipt", "^/_matrix/client/(r0|v3|unstable)/rooms/.*/receipt",
"^/_matrix/client/(r0|v3|unstable)/rooms/.*/read_markers", "^/_matrix/client/(r0|v3|unstable)/rooms/.*/read_markers",
], },
), ),
"to_device": WorkerTemplate( "to_device": WorkerTemplate(
listener_resources=["client", "replication"], listener_resources={"client", "replication"},
endpoint_patterns=["^/_matrix/client/(r0|v3|unstable)/sendToDevice/"], endpoint_patterns={"^/_matrix/client/(r0|v3|unstable)/sendToDevice/"},
), ),
"typing": WorkerTemplate( "typing": WorkerTemplate(
listener_resources=["client", "replication"], listener_resources={"client", "replication"},
endpoint_patterns=["^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/typing"], endpoint_patterns={"^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/typing"},
), ),
} }
@ -406,15 +406,11 @@ def merge_worker_template_configs(
# copy existing_template without any replacements # copy existing_template without any replacements
new_template: WorkerTemplate = dataclasses.replace(existing_template) new_template: WorkerTemplate = dataclasses.replace(existing_template)
# merge the two lists, remove duplicates # add listener resources from the other set
new_template.listener_resources = list( new_template.listener_resources |= to_be_merged_template.listener_resources
set(new_template.listener_resources + to_be_merged_template.listener_resources)
)
# merge the two lists, remove duplicates # add endpoint patterns from the other set
new_template.endpoint_patterns = list( new_template.endpoint_patterns |= to_be_merged_template.endpoint_patterns
set(new_template.endpoint_patterns + to_be_merged_template.endpoint_patterns)
)
# merge dictionaries; the worker name will be replaced later # merge dictionaries; the worker name will be replaced later
new_template.shared_extra_conf = lambda worker_name: { new_template.shared_extra_conf = lambda worker_name: {
@ -444,6 +440,8 @@ def insert_worker_name_for_worker_config(
""" """
dict_to_edit = dataclasses.asdict(existing_template) dict_to_edit = dataclasses.asdict(existing_template)
dict_to_edit["shared_extra_conf"] = existing_template.shared_extra_conf(worker_name) dict_to_edit["shared_extra_conf"] = existing_template.shared_extra_conf(worker_name)
dict_to_edit["endpoint_patterns"] = sorted(existing_template.endpoint_patterns)
dict_to_edit["listener_resources"] = sorted(existing_template.listener_resources)
return dict_to_edit return dict_to_edit
@ -760,7 +758,7 @@ def generate_worker_files(
# Map locations to upstreams (corresponding to worker types) in Nginx # Map locations to upstreams (corresponding to worker types) in Nginx
# but only if we use the appropriate worker type # but only if we use the appropriate worker type
for worker_type in all_worker_types_in_use: for worker_type in all_worker_types_in_use:
for endpoint_pattern in WORKERS_CONFIG[worker_type].endpoint_patterns: for endpoint_pattern in sorted(WORKERS_CONFIG[worker_type].endpoint_patterns):
nginx_locations[endpoint_pattern] = f"http://{worker_type}" nginx_locations[endpoint_pattern] = f"http://{worker_type}"
# For each worker type specified by the user, create config values and write it's # For each worker type specified by the user, create config values and write it's