Draft: /messages investigation scratch pad1

docker/conf/homeserver.yaml

@@ -184,3 +184,21 @@ trusted_key_servers:
 
 password_config:
    enabled: true
+
+
+# foo
+tracing:
+  enabled: true
+  sample_rate: 1
+  jaeger_exporter_config:
+    agent_host_name: host.docker.internal
+    agent_port: 6831
+    # Split UDP packets (UDP_PACKET_MAX_LENGTH is set to 65k in OpenTelemetry)
+    udp_split_oversized_batches: true
+    # If you define a collector, it will communicate directly to the collector,
+    # bypassing the agent
+    #
+    # It does not seem like the agent can keep up with the massive UDP load
+    # (1065 spans in one trace) so lets just use the HTTP collector endpoint
+    # instead which seems to work.
+    collector_endpoint: "http://host.docker.internal:14268/api/traces?format=jaeger.thrift"

pyproject.toml

@@ -119,6 +119,7 @@ signedjson = "^1.1.0"
 service-identity = ">=18.1.0"
 # Twisted 18.9 introduces some logger improvements that the structured
 # logger utilises
+twisted = {git = "https://github.com/twisted/twisted.git", rev = "trunk", extras = ["tls"]}
 treq = ">=15.1"
 # Twisted has required pyopenssl 16.0 since about Twisted 16.6.
 pyOpenSSL = ">=16.0.0"
@@ -182,7 +183,6 @@ idna = { version = ">=2.5", optional = true }
 opentelemetry-api = {version = "^1.11.1", optional = true}
 opentelemetry-sdk = {version = "^1.11.1", optional = true}
 opentelemetry-exporter-jaeger = {version = "^1.11.1", optional = true}
-twisted = {git = "https://github.com/twisted/twisted.git", rev = "trunk"}
 
 [tool.poetry.extras]
 # NB: Packages that should be part of `pip install matrix-synapse[all]` need to be specified

synapse/federation/federation_client.py

@@ -61,7 +61,7 @@
 )
 from synapse.federation.transport.client import SendJoinResponse
 from synapse.http.types import QueryParams
-from synapse.logging.tracing import trace
+from synapse.logging.tracing import tag_args, trace
 from synapse.types import JsonDict, UserID, get_domain_from_id
 from synapse.util.async_helpers import concurrently_execute
 from synapse.util.caches.expiringcache import ExpiringCache
@@ -235,6 +235,7 @@ async def claim_client_keys(
         )
 
     @trace
+    @tag_args
     async def backfill(
         self, dest: str, room_id: str, limit: int, extremities: Collection[str]
     ) -> Optional[List[EventBase]]:
@@ -337,6 +338,8 @@ async def get_pdu_from_destination_raw(
 
         return None
 
+    @trace
+    @tag_args
     async def get_pdu(
         self,
         destinations: Iterable[str],
@@ -474,6 +477,8 @@ async def get_room_state_ids(
 
         return state_event_ids, auth_event_ids
 
+    @trace
+    @tag_args
     async def get_room_state(
         self,
         destination: str,
@@ -533,6 +538,7 @@ async def get_room_state(
 
         return valid_state_events, valid_auth_events
 
+    @trace
     async def _check_sigs_and_hash_and_fetch(
         self,
         origin: str,

synapse/handlers/federation.py

@@ -59,7 +59,7 @@
 from synapse.federation.federation_client import InvalidResponseError
 from synapse.http.servlet import assert_params_in_dict
 from synapse.logging.context import nested_logging_context
-from synapse.logging.tracing import trace
+from synapse.logging.tracing import set_attribute, trace
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.module_api import NOT_SPAM
 from synapse.replication.http.federation import (
@@ -319,7 +319,8 @@ async def _maybe_backfill_inner(
         # attempting to paginate before backfill reached the visible history.
 
         extremities_to_request: List[str] = []
-        for bp in sorted_backfill_points:
+        for i, bp in enumerate(sorted_backfill_points):
+            set_attribute("backfill_point" + str(i), str(bp))
             if len(extremities_to_request) >= 5:
                 break
 

synapse/handlers/federation_event.py

@@ -59,7 +59,13 @@
 from synapse.events.snapshot import EventContext
 from synapse.federation.federation_client import InvalidResponseError
 from synapse.logging.context import nested_logging_context
-from synapse.logging.tracing import trace
+from synapse.logging.tracing import (
+    SynapseTags,
+    set_attribute,
+    start_active_span,
+    tag_args,
+    trace,
+)
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.replication.http.devices import ReplicationUserDevicesResyncRestServlet
 from synapse.replication.http.federation import (
@@ -410,6 +416,7 @@ async def check_join_restrictions(
             prev_member_event,
         )
 
+    @trace
     async def process_remote_join(
         self,
         origin: str,
@@ -715,7 +722,7 @@ async def _get_missing_events_for_pdu(
 
     @trace
     async def _process_pulled_events(
-        self, origin: str, events: Iterable[EventBase], backfilled: bool
+        self, origin: str, events: List[EventBase], backfilled: bool
     ) -> None:
         """Process a batch of events we have pulled from a remote server
 
@@ -730,6 +737,11 @@ async def _process_pulled_events(
             backfilled: True if this is part of a historical batch of events (inhibits
                 notification to clients, and validation of device keys.)
         """
+        set_attribute(
+            SynapseTags.FUNC_ARG_PREFIX + f"event_ids ({len(events)})",
+            str([event.event_id for event in events]),
+        )
+        set_attribute(SynapseTags.FUNC_ARG_PREFIX + "backfilled", str(backfilled))
         logger.debug(
             "processing pulled backfilled=%s events=%s",
             backfilled,
@@ -753,6 +765,7 @@ async def _process_pulled_events(
                 await self._process_pulled_event(origin, ev, backfilled=backfilled)
 
     @trace
+    @tag_args
     async def _process_pulled_event(
         self, origin: str, event: EventBase, backfilled: bool
     ) -> None:
@@ -854,6 +867,7 @@ async def _process_pulled_event(
             else:
                 raise
 
+    @trace
     async def _compute_event_context_with_maybe_missing_prevs(
         self, dest: str, event: EventBase
     ) -> EventContext:
@@ -970,6 +984,8 @@ async def _compute_event_context_with_maybe_missing_prevs(
             event, state_ids_before_event=state_map, partial_state=partial_state
         )
 
+    @trace
+    @tag_args
     async def _get_state_ids_after_missing_prev_event(
         self,
         destination: str,
@@ -1112,6 +1128,8 @@ async def _get_state_ids_after_missing_prev_event(
 
         return state_map
 
+    @trace
+    @tag_args
     async def _get_state_and_persist(
         self, destination: str, room_id: str, event_id: str
     ) -> None:
@@ -1133,6 +1151,7 @@ async def _get_state_and_persist(
                 destination=destination, room_id=room_id, event_ids=(event_id,)
             )
 
+    @trace
     async def _process_received_pdu(
         self,
         origin: str,
@@ -1283,6 +1302,7 @@ async def _resync_device(self, sender: str) -> None:
         except Exception:
             logger.exception("Failed to resync device for %s", sender)
 
+    @trace
     async def _handle_marker_event(self, origin: str, marker_event: EventBase) -> None:
         """Handles backfilling the insertion event when we receive a marker
         event that points to one.
@@ -1414,6 +1434,8 @@ async def backfill_event_id(
 
         return event_from_response
 
+    @trace
+    @tag_args
     async def _get_events_and_persist(
         self, destination: str, room_id: str, event_ids: Collection[str]
     ) -> None:
@@ -1459,6 +1481,7 @@ async def get_event(event_id: str) -> None:
         logger.info("Fetched %i events of %i requested", len(events), len(event_ids))
         await self._auth_and_persist_outliers(room_id, events)
 
+    @trace
     async def _auth_and_persist_outliers(
         self, room_id: str, events: Iterable[EventBase]
     ) -> None:
@@ -1477,6 +1500,12 @@ async def _auth_and_persist_outliers(
         """
         event_map = {event.event_id: event for event in events}
 
+        event_ids = event_map.keys()
+        set_attribute(
+            SynapseTags.FUNC_ARG_PREFIX + f"event_ids ({len(event_ids)})",
+            str(event_ids),
+        )
+
         # filter out any events we have already seen. This might happen because
         # the events were eagerly pushed to us (eg, during a room join), or because
         # another thread has raced against us since we decided to request the event.
@@ -1593,6 +1622,7 @@ async def prep(event: EventBase) -> None:
             backfilled=True,
         )
 
+    @trace
     async def _check_event_auth(
         self, origin: Optional[str], event: EventBase, context: EventContext
     ) -> None:
@@ -1631,6 +1661,9 @@ async def _check_event_auth(
         claimed_auth_events = await self._load_or_fetch_auth_events_for_event(
             origin, event
         )
+        set_attribute(
+            "claimed_auth_events", str([ev.event_id for ev in claimed_auth_events])
+        )
 
         # ... and check that the event passes auth at those auth events.
         # https://spec.matrix.org/v1.3/server-server-api/#checks-performed-on-receipt-of-a-pdu:
@@ -1728,6 +1761,7 @@ async def _check_event_auth(
             )
             context.rejected = RejectedReason.AUTH_ERROR
 
+    @trace
     async def _maybe_kick_guest_users(self, event: EventBase) -> None:
         if event.type != EventTypes.GuestAccess:
             return
@@ -1935,6 +1969,8 @@ async def _load_or_fetch_auth_events_for_event(
         # instead we raise an AuthError, which will make the caller ignore it.
         raise AuthError(code=HTTPStatus.FORBIDDEN, msg="Auth events could not be found")
 
+    @trace
+    @tag_args
     async def _get_remote_auth_chain_for_event(
         self, destination: str, room_id: str, event_id: str
     ) -> None:
@@ -1963,6 +1999,7 @@ async def _get_remote_auth_chain_for_event(
 
         await self._auth_and_persist_outliers(room_id, remote_auth_events)
 
+    @trace
     async def _run_push_actions_and_persist_event(
         self, event: EventBase, context: EventContext, backfilled: bool = False
     ) -> None:
@@ -2071,8 +2108,13 @@ async def persist_events_and_notify(
                     self._message_handler.maybe_schedule_expiry(event)
 
             if not backfilled:  # Never notify for backfilled events
-                for event in events:
-                    await self._notify_persisted_event(event, max_stream_token)
+                with start_active_span("notify_persisted_events"):
+                    set_attribute(
+                        SynapseTags.FUNC_ARG_PREFIX + f"event_ids ({len(events)})",
+                        str([ev.event_id for ev in events]),
+                    )
+                    for event in events:
+                        await self._notify_persisted_event(event, max_stream_token)
 
             return max_stream_token.stream