[WIP] Upgrade to Synapse 1.18

build/purger/purger.py

@@ -83,7 +83,9 @@ def purge(
 
     api = GMatrixHttpApi(server)
     try:
-        response = api.login("m.login.password", user=username, password=password)
+        response = api.login(
+            "m.login.password", user=username, password=password, device_id="purger"
+        )
         api.token = response["access_token"]
     except (MatrixError, KeyError) as ex:
         click.secho(f"Could not log in to server {server}: {ex}")
@@ -123,7 +125,18 @@ def purge(
                 # In case an empty env var is set
                 url_known_federation_servers = DEFAULT_MATRIX_KNOWN_SERVERS[Environment.PRODUCTION]
             # fetch remote whiltelist
-            remote_whitelist = yaml.safe_load(requests.get(url_known_federation_servers).text)
+            try:
+                remote_whitelist = json.loads(requests.get(url_known_federation_servers).text)[
+                    "all_servers"
+                ]
+            except (requests.RequestException, JSONDecodeError, KeyError) as ex:
+                click.secho(
+                    f"Error while fetching whitelist: {ex!r}. "
+                    f"Ignoring, containers will be restarted.",
+                    err=True,
+                )
+                # An empty whitelist will cause the container to be restarted
+                remote_whitelist = []
 
             client = docker.from_env()  # pylint: disable=no-member
             for container in client.containers.list():
@@ -143,16 +156,9 @@ def purge(
                         continue
 
                     click.secho(f"Whitelist changed. Restarting. new_list={remote_whitelist!r}")
-                except (
-                    KeyError,
-                    IndexError,
-                    requests.RequestException,
-                    yaml.scanner.ScannerError,
-                ) as ex:
+                except (KeyError, IndexError) as ex:
                     click.secho(
-                        f"An error ocurred while fetching whitelists: {ex!r}\n"
-                        "Restarting anyway",
-                        err=True,
+                        f"Error fetching container status: {ex!r}. Restarting anyway.", err=True,
                     )
                 # restart container
                 container.restart(timeout=30)

build/room_ensurer/room_ensurer.py

@@ -26,8 +26,6 @@
 
 patch_all()  # isort:skip
 
-from raiden.utils.datastructures import merge_dict
-
 import json
 import os
 import sys
@@ -42,7 +40,6 @@
 import gevent
 from eth_utils import encode_hex, to_normalized_address
 from matrix_client.errors import MatrixError
-from raiden_contracts.utils.type_aliases import ChainID
 from structlog import get_logger
 
 from raiden.constants import (
@@ -51,14 +48,16 @@
     PATH_FINDING_BROADCASTING_ROOM,
     Environment,
     Networks,
+    ServerListType,
 )
 from raiden.log_config import configure_logging
 from raiden.network.transport.matrix import make_room_alias
 from raiden.network.transport.matrix.client import GMatrixHttpApi
 from raiden.settings import DEFAULT_MATRIX_KNOWN_SERVERS
 from raiden.tests.utils.factories import make_signer
 from raiden.utils.cli import get_matrix_servers
-
+from raiden.utils.datastructures import merge_dict
+from raiden_contracts.utils.type_aliases import ChainID
 
 ENV_KEY_KNOWN_SERVERS = "URL_KNOWN_FEDERATION_SERVERS"
 
@@ -89,11 +88,11 @@ class RoomInfo:
 
 class RoomEnsurer:
     def __init__(
-            self,
-            username: str,
-            password: str,
-            own_server_name: str,
-            known_servers_url: Optional[str] = None,
+        self,
+        username: str,
+        password: str,
+        own_server_name: str,
+        known_servers_url: Optional[str] = None,
     ):
         self._username = username
         self._password = password
@@ -104,7 +103,9 @@ def __init__(
 
         self._known_servers: Dict[str, str] = {
             urlparse(server_url).netloc: server_url
-            for server_url in get_matrix_servers(known_servers_url)
+            for server_url in get_matrix_servers(
+                known_servers_url, server_list_type=ServerListType.ALL_SERVERS
+            )
         }
         if not self._known_servers:
             raise RuntimeError(f"No known servers found from list at {known_servers_url}.")
@@ -214,7 +215,7 @@ def _ensure_room_for_network(self, network: Networks, alias_fragment: str) -> No
         self._ensure_admin_power_levels(room_infos[self._own_server_name])
 
     def _join_and_alias_room(
-            self, first_server_room_alias: str, own_server_room_alias: str
+        self, first_server_room_alias: str, own_server_room_alias: str
     ) -> None:
         response = self._own_api.join_room(first_server_room_alias)
         own_room_id = response.get("room_id")
@@ -291,12 +292,16 @@ def _ensure_admin_power_levels(self, room_info: RoomInfo) -> None:
             return
 
         if own_user not in current_power_levels["users"]:
-            log.warning(f"{own_user} has not been granted administrative power levels yet. Doing nothing.")
+            log.warning(
+                f"{own_user} has not been granted administrative power levels yet. Doing nothing."
+            )
             return
 
         # the supposed power level dict could be just a subset of the current
         # because providers who left cannot be removed from other admins
-        if set(supposed_power_levels["users"].keys()).issubset(set(current_power_levels["users"].keys())):
+        if set(supposed_power_levels["users"].keys()).issubset(
+            set(current_power_levels["users"].keys())
+        ):
             log.debug(f"Power levels are up to date. Doing nothing.")
             return
 
@@ -338,7 +343,9 @@ def _connect(self, server_name: str, server_url: str) -> Tuple[str, GMatrixHttpA
             username = str(to_normalized_address(signer.address))
             password = encode_hex(signer.sign(server_name.encode()))
 
-        response = api.login("m.login.password", user=username, password=password)
+        response = api.login(
+            "m.login.password", user=username, password=password, device_id="room_ensurer"
+        )
         api.token = response["access_token"]
         log.debug("Connected", server=server_name)
         return server_name, api

build/synapse/Dockerfile

@@ -13,7 +13,7 @@ ARG SYNAPSE_VERSION
 
 RUN \
     python -m venv /synapse-venv && \
-    /synapse-venv/bin/pip install "matrix-synapse[postgres]==${SYNAPSE_VERSION}"
+    /synapse-venv/bin/pip install "matrix-synapse[postgres,redis]==${SYNAPSE_VERSION}"
 
 RUN /synapse-venv/bin/pip install psycopg2 coincurve pycryptodome twisted>=20.3.0
 
@@ -24,14 +24,12 @@ COPY render_config_template.py /bin/
 COPY --from=RAIDEN /known_servers.default.txt /
 
 ENTRYPOINT ["/bin/synapse-entrypoint.sh"]
-CMD ["/synapse-venv/bin/python", "-m", "synapse.app.homeserver", "--config-path", "/config/synapse.yaml"]
+CMD ["synapse"]
 
 # HTTP
 EXPOSE 8008
 # HTTP metrics
 EXPOSE 9101
-# TCP replication
-EXPOSE 9092
 # HTTP replication
 EXPOSE 9093
 

build/synapse/render_config_template.py

@@ -2,16 +2,23 @@
 import os
 import random
 import string
+from json.decoder import JSONDecodeError
 from pathlib import Path
 from typing import Optional
 from urllib.error import URLError
 from urllib.request import urlopen
 
-PATH_CONFIG = Path("/config/synapse.yaml")
-PATH_CONFIG_TEMPLATE = Path("/config/synapse.template.yaml")
+import click
+from docker import Client
+
+PATH_CONFIG_SYNAPSE = Path("/config/synapse.yaml")
+PATH_CONFIG_TEMPLATE_SYNAPSE = Path("/config/synapse.template.yaml")
+PATH_CONFIG_WORKER_BASE = Path("/config/workers/")
+PATH_CONFIG_TEMPLATE_WORKER = PATH_CONFIG_WORKER_BASE.joinpath("worker.template.yaml")
+
 PATH_MACAROON_KEY = Path("/data/keys/macaroon.key")
 PATH_ADMIN_USER_CREDENTIALS = Path("/config/admin_user_cred.json")
-PATH_KNOWN_FEDERATION_SERVERS = Path("/data/known_federation_servers.yaml")
+PATH_KNOWN_FEDERATION_SERVERS = Path("/data/known_federation_servers.json")
 PATH_WELL_KNOWN_FILE = Path("/data_well_known/server")
 
 # This file gets created during docker build from the given Raiden version
@@ -36,7 +43,13 @@ def get_known_federation_servers(url_known_federation_servers: Optional[str]) ->
     try:
         resp = urlopen(url_known_federation_servers)
         if 200 <= resp.code < 300:
-            PATH_KNOWN_FEDERATION_SERVERS.write_text(resp.read().decode())
+            try:
+                known_servers = json.loads(resp.read().decode())
+                PATH_KNOWN_FEDERATION_SERVERS.write_text(
+                    "".join(f"- {server}\n" for server in known_servers["all_servers"])
+                )
+            except (JSONDecodeError, KeyError):
+                print("Error loading known servers list:", resp.code, resp.read().decode())
         else:
             print("Error fetching known servers list:", resp.code, resp.read().decode())
     except URLError as ex:
@@ -47,13 +60,13 @@ def get_known_federation_servers(url_known_federation_servers: Optional[str]) ->
 
 
 def render_synapse_config(server_name: str, url_known_federation_servers: Optional[str]) -> None:
-    template_content = PATH_CONFIG_TEMPLATE.read_text()
+    template_content = PATH_CONFIG_TEMPLATE_SYNAPSE.read_text()
     rendered_config = string.Template(template_content).substitute(
         MACAROON_KEY=get_macaroon_key(),
         SERVER_NAME=server_name,
         KNOWN_SERVERS=get_known_federation_servers(url_known_federation_servers),
     )
-    PATH_CONFIG.write_text(rendered_config)
+    PATH_CONFIG_SYNAPSE.write_text(rendered_config)
 
 
 def render_well_known_file(server_name: str) -> None:
@@ -76,7 +89,34 @@ def generate_admin_user_credentials() -> None:
     )
 
 
+def render_worker_config(type_: str) -> Path:
+    # Fetch compose scale index
+    client = Client.from_env()
+    own_container_labels = next(
+        container["Labels"]
+        for container in client.containers()
+        if container["Id"].startswith(os.environ["HOSTNAME"])
+    )
+    own_container_index = (
+        int(own_container_labels.get("com.docker.compose.container-number"), 0) - 1
+    )
+
+    template_content = PATH_CONFIG_TEMPLATE_WORKER.read_text()
+    rendered_content = string.Template(template_content).substitute(
+        WORKER_APP=type_, WORKER_INDEX=own_container_index
+    )
+    target_file = PATH_CONFIG_WORKER_BASE.joinpath(f"{type_}_{own_container_index}.yaml")
+    target_file.write_text(rendered_content)
+    return target_file
+
+
+@click.group()
 def main() -> None:
+    pass
+
+
+@main.command()
+def synapse() -> None:
     url_known_federation_servers = os.environ.get("URL_KNOWN_FEDERATION_SERVERS")
     server_name = os.environ["SERVER_NAME"]
 
@@ -87,5 +127,14 @@ def main() -> None:
     generate_admin_user_credentials()
 
 
+@main.command()
+@click.option(
+    "--type", "type_", type=click.Choice(["generic_worker", "federation_sender", "user_dir"])
+)
+def worker(type_) -> None:
+    target_file = render_worker_config(type_)
+    print(str(target_file))
+
+
 if __name__ == "__main__":
     main()

build/synapse/synapse-entrypoint.sh

@@ -4,6 +4,21 @@
 mkdir -p /data/log
 mkdir -p /data/keys
 
-/synapse-venv/bin/python /bin/render_config_template.py
-/synapse-venv/bin/python -m synapse.app.homeserver --config-path /config/synapse.yaml --generate-keys
-exec "$@"
+TYPE="$1"
+shift
+
+if [[ $TYPE == 'worker' ]]; then
+  WORKER="$1"
+  shift
+  CONFIG_PATH=$(/synapse-venv/bin/python /bin/render_config_template.py "$TYPE" --type "$WORKER")
+
+  /synapse-venv/bin/python -m "synapse.app.${WORKER}" --config-path /config/synapse.yaml --config-path "${CONFIG_PATH}"
+elif [[ $TYPE == 'synapse' ]]; then
+  /synapse-venv/bin/python /bin/render_config_template.py "$TYPE"
+  /synapse-venv/bin/python -m synapse.app.homeserver --config-path /config/synapse.yaml --generate-keys
+
+  /synapse-venv/bin/python -m synapse.app.homeserver --config-path /config/synapse.yaml --config-path /config/workers/homeserver.yaml
+else
+  echo "Unknown run type: $TYPE"
+  exit 1
+fi

config/metrics_db/queries.yaml

@@ -0,0 +1,15 @@
+pg_table:
+  query: "SELECT current_database() as datname, relname AS table_name, c.reltuples AS row_estimate, pg_total_relation_size(c.oid) AS size_bytes FROM pg_class c LEFT JOIN pg_namespace n ON n.oid = c.relnamespace WHERE relkind = 'r' AND nspname = 'public' ORDER BY table_name;"
+  metrics:
+    - datname:
+        usage: "LABEL"
+        description: "Database name"
+    - table_name:
+        usage: "LABEL"
+        description: "Table name"
+    - row_estimate:
+        usage: "GAUGE"
+        description: "Estimated row count"
+    - size_bytes:
+        usage: "GAUGE"
+        description: "Table size"

config/synapse/synapse.template.yaml

@@ -25,18 +25,16 @@ listeners:
     bind_addresses: ['0.0.0.0']
     type: metrics
 
-  # TCP replication
-  - port: 9092
-    bind_address: '0.0.0.0'
-    type: replication
-
   # HTTP replication
   - port: 9093
     bind_address: '0.0.0.0'
     type: http
     resources:
       - names: [replication]
 
+redis:
+  enabled: true
+  host: redis
 
 event_cache_size: "20K"
 
@@ -48,16 +46,16 @@ log_config: "/config/synapse.log.config"
 # Handled by the federation_sender worker
 send_federation: False
 # Handled by the user_directory worker
-update_user_directory: false
+update_user_directory: False
 
 
 ## Ratelimiting
 
 rc_message:
   # Number of messages a client can send per second on average
-  per_second: 15
+  per_second: 30
   # Number of message a client can send before being throttled
-  burst_count: 100
+  burst_count: 200
 
 rc_login:
   address:
@@ -75,10 +73,10 @@ rc_federation:
   window_size: 1000
   # The number of federation requests from a single server in a window
   # before the server will delay processing the request.
-  sleep_limit: 50
+  sleep_limit: 250
   # The duration in milliseconds to delay processing events from
   # remote servers by if they go over the sleep limit.
-  sleep_delay: 250
+  sleep_delay: 125
   # The maximum number of concurrent federation requests allowed
   # from a single server
   reject_limit: 50