Test that HAProxy routes correctly to Synapse workers

Author: benbz

charts/matrix-stack/ci/fragments/synapse-pytest-self-extras.yaml

@@ -20,10 +20,16 @@ synapse:
     # A non-HTTP worker & a stream writer
     event-persister:
       enabled: true
+    # initial-synchrotron & synchrotron have non-trivial routing behaviour
+    initial-synchrotron:
+      enabled: true
     # A standard HTTP worker
     sliding-sync:
       replicas: 2
       enabled: true
+    # initial-synchrotron & synchrotron have non-trivial routing behaviour
+    synchrotron:
+      enabled: true
     # Media repo is fairly distinct from other workers
     media-repository:
       enabled: true

charts/matrix-stack/ci/pytest-synapse-values.yaml

@@ -58,12 +58,18 @@ synapse:
     # A non-HTTP worker & a stream writer
     event-persister:
       enabled: true
+    # initial-synchrotron & synchrotron have non-trivial routing behaviour
+    initial-synchrotron:
+      enabled: true
     # Media repo is fairly distinct from other workers
     media-repository:
       enabled: true
     # A standard HTTP worker
     sliding-sync:
       enabled: true
       replicas: 2
+    # initial-synchrotron & synchrotron have non-trivial routing behaviour
+    synchrotron:
+      enabled: true
 wellKnownDelegation:
   enabled: false

tests/integration/test_synapse.py

@@ -5,13 +5,17 @@
 import hashlib
 from pathlib import Path
 
+import aiohttp
 import pyhelm3
 import pytest
+from aiohttp_retry import RetryClient
+from lightkube import AsyncClient
+from lightkube.resources.core_v1 import Pod
 
 from .fixtures import ESSData, User
 from .lib.helpers import deploy_with_values_patch, get_deployment_marker
 from .lib.synapse import assert_downloaded_content, download_media, upload_media
-from .lib.utils import KubeCtl, aiohttp_client, aiohttp_get_json, aiohttp_post_json, value_file_has
+from .lib.utils import KubeCtl, aiohttp_client, aiohttp_get_json, aiohttp_post_json, retry_options, value_file_has
 
 
 @pytest.mark.skipif(value_file_has("synapse.enabled", False), reason="Synapse not deployed")
@@ -35,7 +39,9 @@ async def test_synapse_can_access_client_api(
 @pytest.mark.skipif(value_file_has("synapse.enabled", False), reason="Synapse not deployed")
 @pytest.mark.parametrize("users", [(User(name="sliding-sync-user"),)], indirect=True)
 @pytest.mark.asyncio_cooperative
-async def test_simplified_sliding_sync_syncs(ssl_context, users, generated_data: ESSData):
+async def test_simplified_sliding_sync_syncs(ingress_ready, ssl_context, users, generated_data: ESSData):
+    await ingress_ready("synapse")
+
     access_token = users[0].access_token
 
     sync_result = await aiohttp_post_json(
@@ -48,6 +54,79 @@ async def test_simplified_sliding_sync_syncs(ssl_context, users, generated_data:
     assert "pos" in sync_result
 
 
+@pytest.mark.skipif(value_file_has("synapse.enabled", False), reason="Synapse not deployed")
+@pytest.mark.asyncio_cooperative
+async def test_routes_to_synapse_workers_correctly(
+    ingress_ready, kube_client: AsyncClient, ssl_context, generated_data: ESSData
+):
+    await ingress_ready("synapse")
+
+    main_backend = "main/main"
+    if value_file_has("synapse.workers.sliding-sync.enabled", True):
+        sliding_sync_backend = "sliding-sync/sliding-sync"
+    else:
+        sliding_sync_backend = main_backend
+
+    if value_file_has("synapse.workers.synchrotron.enabled", True):
+        synchrotron_backend = "synchrotron/synchrotron"
+    else:
+        synchrotron_backend = main_backend
+
+    if value_file_has("synapse.workers.initial-synchrotron.enabled", True):
+        initial_synchrotron_backend = "initial-synchrotron/initial-sync"
+    else:
+        initial_synchrotron_backend = synchrotron_backend
+
+    # We don't care about any of these succeeding, only that the requests are made and HAProxy dispatches correctly
+    # So no auth required and parameters can be made up
+    paths_to_backends = {
+        # initial-synchrotron
+        "/_matrix/client/v3/initialSync": initial_synchrotron_backend,
+        "/_matrix/client/v3/rooms/aroomid/initialSync": initial_synchrotron_backend,
+        "/_matrix/client/v3/sync?full_state=true": initial_synchrotron_backend,
+        "/_matrix/client/v3/sync": initial_synchrotron_backend,
+        "/_matrix/client/v3/events": initial_synchrotron_backend,
+        # synchrotron
+        "/_matrix/client/v3/sync?since=recently": synchrotron_backend,
+        "/_matrix/client/v3/events?from=recently": synchrotron_backend,
+        # sliding-sync
+        "/_matrix/client/unstable/org.matrix.simplified_msc3575/sync": sliding_sync_backend,
+        # Would be client-reader but not configured in tests
+        "/_matrix/client/versions": main_backend,
+    }
+
+    for path in paths_to_backends:
+        host = f"synapse.{generated_data.server_name}"
+        url = f"https://127.0.0.1{path}"
+        async with (
+            aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl_context)) as session,
+            RetryClient(session, retry_options=retry_options, raise_for_status=False) as client,
+            client.get(url, headers={"Host": host}, server_hostname=host) as response,
+        ):
+            # 200 for no auth required (/_matrix/client/versions)
+            # 401 for most paths that require auth
+            # 405 for sliding sync that doesn't do GET
+            assert response.status in [200, 401, 405], f"{path} had an unexpected status"
+
+    http_log_lines = []
+    async for haproxy_pod in kube_client.list(
+        Pod, namespace=generated_data.ess_namespace, labels={"app.kubernetes.io/name": "haproxy"}
+    ):
+        assert haproxy_pod.metadata
+        assert haproxy_pod.metadata.name
+        assert haproxy_pod.metadata.namespace
+        async for log_line in kube_client.log(haproxy_pod.metadata.name, namespace=haproxy_pod.metadata.namespace):
+            if "HTTP/1.1" in log_line:
+                http_log_lines.append(log_line)
+
+    for path, backend in paths_to_backends.items():
+        matching_lines = [line for line in http_log_lines if f"GET {path} HTTP/1.1" in line]
+
+        assert len(matching_lines) > 0, f"Requests for {path} did not appear in the HAProxy logs"
+        for matching_line in matching_lines:
+            assert f"synapse-http-in synapse-{backend}" in matching_line, f"{path} was routed unexpectedly"
+
+
 @pytest.mark.skipif(value_file_has("synapse.enabled", False), reason="Synapse not deployed")
 @pytest.mark.parametrize("users", [(User(name="media-upload-unauth"),)], indirect=True)
 @pytest.mark.asyncio_cooperative