From 86f39b04a61984de0076b32b1f80b41906dbacd7 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 16 May 2025 17:10:22 -0500 Subject: [PATCH] Partial `Counter` update --- synapse/appservice/api.py | 72 ++++++----- synapse/federation/federation_client.py | 17 ++- synapse/federation/federation_server.py | 61 +++++---- synapse/federation/sender/__init__.py | 28 +++-- .../sender/per_destination_queue.py | 28 +++-- synapse/handlers/appservice.py | 10 +- synapse/handlers/auth.py | 19 +-- synapse/handlers/federation_event.py | 76 ++++++------ synapse/handlers/presence.py | 117 ++++++++++++------ 9 files changed, 256 insertions(+), 172 deletions(-) diff --git a/synapse/appservice/api.py b/synapse/appservice/api.py index 19322471dc..008873450f 100644 --- a/synapse/appservice/api.py +++ b/synapse/appservice/api.py @@ -56,33 +56,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -sent_transactions_counter = Counter( - "synapse_appservice_api_sent_transactions", - "Number of /transactions/ requests sent", - ["service"], -) - -failed_transactions_counter = Counter( - "synapse_appservice_api_failed_transactions", - "Number of /transactions/ requests that failed to send", - ["service"], -) - -sent_events_counter = Counter( - "synapse_appservice_api_sent_events", "Number of events sent to the AS", ["service"] -) - -sent_ephemeral_counter = Counter( - "synapse_appservice_api_sent_ephemeral", - "Number of ephemeral events sent to the AS", - ["service"], -) - -sent_todevice_counter = Counter( - "synapse_appservice_api_sent_todevice", - "Number of todevice messages sent to the AS", - ["service"], -) HOUR_IN_MS = 60 * 60 * 1000 @@ -133,6 +106,41 @@ class ApplicationServiceApi(SimpleHttpClient): hs.get_clock(), "as_protocol_meta", timeout_ms=HOUR_IN_MS ) + self.sent_transactions_counter = Counter( + "synapse_appservice_api_sent_transactions", + "Number of /transactions/ requests sent", + ["service"], + registry=hs.metrics_collector_registry, + ) + + self.failed_transactions_counter = Counter( + "synapse_appservice_api_failed_transactions", + "Number of /transactions/ requests that failed to send", + ["service"], + registry=hs.metrics_collector_registry, + ) + + self.sent_events_counter = Counter( + "synapse_appservice_api_sent_events", + "Number of events sent to the AS", + ["service"], + registry=hs.metrics_collector_registry, + ) + + self.sent_ephemeral_counter = Counter( + "synapse_appservice_api_sent_ephemeral", + "Number of ephemeral events sent to the AS", + ["service"], + registry=hs.metrics_collector_registry, + ) + + self.sent_todevice_counter = Counter( + "synapse_appservice_api_sent_todevice", + "Number of todevice messages sent to the AS", + ["service"], + registry=hs.metrics_collector_registry, + ) + def _get_headers(self, service: "ApplicationService") -> Dict[bytes, List[bytes]]: """This makes sure we have always the auth header and opentracing headers set.""" @@ -395,10 +403,10 @@ class ApplicationServiceApi(SimpleHttpClient): service.url, [event.get("event_id") for event in events], ) - sent_transactions_counter.labels(service.id).inc() - sent_events_counter.labels(service.id).inc(len(serialized_events)) - sent_ephemeral_counter.labels(service.id).inc(len(ephemeral)) - sent_todevice_counter.labels(service.id).inc(len(to_device_messages)) + self.sent_transactions_counter.labels(service.id).inc() + self.sent_events_counter.labels(service.id).inc(len(serialized_events)) + self.sent_ephemeral_counter.labels(service.id).inc(len(ephemeral)) + self.sent_todevice_counter.labels(service.id).inc(len(to_device_messages)) return True except CodeMessageException as e: logger.warning( @@ -417,7 +425,7 @@ class ApplicationServiceApi(SimpleHttpClient): ex.args, exc_info=logger.isEnabledFor(logging.DEBUG), ) - failed_transactions_counter.labels(service.id).inc() + self.failed_transactions_counter.labels(service.id).inc() return False async def claim_client_keys( diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 9fc5b70e9a..73106f861a 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -84,8 +84,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -sent_queries_counter = Counter("synapse_federation_client_sent_queries", "", ["type"]) - PDU_RETRY_TIME_MS = 1 * 60 * 1000 @@ -167,6 +165,13 @@ class FederationClient(FederationBase): reset_expiry_on_get=False, ) + self.sent_queries_counter = Counter( + "synapse_federation_client_sent_queries", + "", + ["type"], + registry=hs.metrics_collector_registry, + ) + def _clear_tried_cache(self) -> None: """Clear pdu_destination_tried cache""" now = self._clock.time_msec() @@ -206,7 +211,7 @@ class FederationClient(FederationBase): Returns: The JSON object from the response """ - sent_queries_counter.labels(query_type).inc() + self.sent_queries_counter.labels(query_type).inc() return await self.transport_layer.make_query( destination, @@ -228,7 +233,7 @@ class FederationClient(FederationBase): Returns: The JSON object from the response """ - sent_queries_counter.labels("client_device_keys").inc() + self.sent_queries_counter.labels("client_device_keys").inc() return await self.transport_layer.query_client_keys( destination, content, timeout ) @@ -239,7 +244,7 @@ class FederationClient(FederationBase): """Query the device keys for a list of user ids hosted on a remote server. """ - sent_queries_counter.labels("user_devices").inc() + self.sent_queries_counter.labels("user_devices").inc() return await self.transport_layer.query_user_devices( destination, user_id, timeout ) @@ -261,7 +266,7 @@ class FederationClient(FederationBase): Returns: The JSON object from the response """ - sent_queries_counter.labels("client_one_time_keys").inc() + self.sent_queries_counter.labels("client_one_time_keys").inc() # Convert the query with counts into a stable and unstable query and check # if attempting to claim more than 1 OTK. diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 2f2c78babc..fde97d7a69 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -105,25 +105,6 @@ TRANSACTION_CONCURRENCY_LIMIT = 10 logger = logging.getLogger(__name__) -received_pdus_counter = Counter("synapse_federation_server_received_pdus", "") - -received_edus_counter = Counter("synapse_federation_server_received_edus", "") - -received_queries_counter = Counter( - "synapse_federation_server_received_queries", "", ["type"] -) - -pdu_process_time = Histogram( - "synapse_federation_server_pdu_process_time", - "Time taken to process an event", -) - -last_pdu_ts_metric = Gauge( - "synapse_federation_last_received_pdu_time", - "The timestamp of the last PDU which was successfully received from the given domain", - labelnames=("server_name",), -) - # The name of the lock to use when process events in a room received over # federation. @@ -185,6 +166,38 @@ class FederationServer(FederationBase): # Whether we have started handling old events in the staging area. self._started_handling_of_staged_events = False + self.received_pdus_counter = Counter( + "synapse_federation_server_received_pdus", + "", + registry=hs.metrics_collector_registry, + ) + + self.received_edus_counter = Counter( + "synapse_federation_server_received_edus", + "", + registry=hs.metrics_collector_registry, + ) + + self.received_queries_counter = Counter( + "synapse_federation_server_received_queries", + "", + ["type"], + registry=hs.metrics_collector_registry, + ) + + self.pdu_process_time = Histogram( + "synapse_federation_server_pdu_process_time", + "Time taken to process an event", + registry=hs.metrics_collector_registry, + ) + + self.last_pdu_ts_metric = Gauge( + "synapse_federation_last_received_pdu_time", + "The timestamp of the last PDU which was successfully received from the given domain", + labelnames=("server_name",), + registry=hs.metrics_collector_registry, + ) + @wrap_as_background_process("_handle_old_staged_events") async def _handle_old_staged_events(self) -> None: """Handle old staged events by fetching all rooms that have staged @@ -424,7 +437,7 @@ class FederationServer(FederationBase): report back to the sending server. """ - received_pdus_counter.inc(len(transaction.pdus)) + self.received_pdus_counter.inc(len(transaction.pdus)) origin_host, _ = parse_server_name(origin) @@ -535,7 +548,7 @@ class FederationServer(FederationBase): ) if newest_pdu_ts and origin in self._federation_metrics_domains: - last_pdu_ts_metric.labels(server_name=origin).set(newest_pdu_ts / 1000) + self.last_pdu_ts_metric.labels(server_name=origin).set(newest_pdu_ts / 1000) return pdu_results @@ -543,7 +556,7 @@ class FederationServer(FederationBase): """Process the EDUs in a received transaction.""" async def _process_edu(edu_dict: JsonDict) -> None: - received_edus_counter.inc() + self.received_edus_counter.inc() edu = Edu( origin=origin, @@ -658,7 +671,7 @@ class FederationServer(FederationBase): async def on_query_request( self, query_type: str, args: Dict[str, str] ) -> Tuple[int, Dict[str, Any]]: - received_queries_counter.labels(query_type).inc() + self.received_queries_counter.labels(query_type).inc() resp = await self.registry.on_query(query_type, args) return 200, resp @@ -1299,7 +1312,7 @@ class FederationServer(FederationBase): origin, event.event_id ) if received_ts is not None: - pdu_process_time.observe( + self.pdu_process_time.observe( (self._clock.time_msec() - received_ts) / 1000 ) diff --git a/synapse/federation/sender/__init__.py b/synapse/federation/sender/__init__.py index b95b3c629d..27654b12f7 100644 --- a/synapse/federation/sender/__init__.py +++ b/synapse/federation/sender/__init__.py @@ -186,15 +186,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -sent_pdus_destination_dist_count = Counter( - "synapse_federation_client_sent_pdu_destinations_count", - "Number of PDUs queued for sending to one or more destinations", -) - -sent_pdus_destination_dist_total = Counter( - "synapse_federation_client_sent_pdu_destinations", - "Total number of PDUs queued for sending across all destinations", -) # Time (in s) to wait before trying to wake up destinations that have # catch-up outstanding. @@ -397,6 +388,7 @@ class FederationSender(AbstractFederationSender): for d in self._per_destination_queues.values() if d.transmission_loop_running ), + registry=hs.metrics_collector_registry, ) LaterGauge( @@ -406,6 +398,7 @@ class FederationSender(AbstractFederationSender): lambda: sum( d.pending_pdu_count() for d in self._per_destination_queues.values() ), + registry=hs.metrics_collector_registry, ) LaterGauge( "synapse_federation_transaction_queue_pending_edus", @@ -414,6 +407,19 @@ class FederationSender(AbstractFederationSender): lambda: sum( d.pending_edu_count() for d in self._per_destination_queues.values() ), + registry=hs.metrics_collector_registry, + ) + + self.sent_pdus_destination_dist_count = Counter( + "synapse_federation_client_sent_pdu_destinations_count", + "Number of PDUs queued for sending to one or more destinations", + registry=hs.metrics_collector_registry, + ) + + self.sent_pdus_destination_dist_total = Counter( + "synapse_federation_client_sent_pdu_destinations", + "Total number of PDUs queued for sending across all destinations", + registry=hs.metrics_collector_registry, ) self._is_processing = False @@ -710,8 +716,8 @@ class FederationSender(AbstractFederationSender): if not destinations: return - sent_pdus_destination_dist_total.inc(len(destinations)) - sent_pdus_destination_dist_count.inc() + self.sent_pdus_destination_dist_total.inc(len(destinations)) + self.sent_pdus_destination_dist_count.inc() assert pdu.internal_metadata.stream_ordering diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py index b3f65e8237..165ac3dfa1 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/synapse/federation/sender/per_destination_queue.py @@ -55,17 +55,6 @@ MAX_EDUS_PER_TRANSACTION = 100 logger = logging.getLogger(__name__) -sent_edus_counter = Counter( - "synapse_federation_client_sent_edus", "Total number of EDUs successfully sent" -) - -sent_edus_by_type = Counter( - "synapse_federation_client_sent_edus_by_type", - "Number of sent EDUs successfully sent, by event type", - ["type"], -) - - # If the retry interval is larger than this then we enter "catchup" mode CATCHUP_RETRY_INTERVAL = 60 * 60 * 1000 @@ -164,6 +153,19 @@ class PerDestinationQueue: # stream_id of last successfully sent device list update. self._last_device_list_stream_id = 0 + self.sent_edus_counter = Counter( + "synapse_federation_client_sent_edus", + "Total number of EDUs successfully sent", + registry=hs.metrics_collector_registry, + ) + + self.sent_edus_by_type = Counter( + "synapse_federation_client_sent_edus_by_type", + "Number of sent EDUs successfully sent, by event type", + ["type"], + registry=hs.metrics_collector_registry, + ) + def __str__(self) -> str: return "PerDestinationQueue[%s]" % self._destination @@ -361,9 +363,9 @@ class PerDestinationQueue: ) sent_transactions_counter.inc() - sent_edus_counter.inc(len(pending_edus)) + self.sent_edus_counter.inc(len(pending_edus)) for edu in pending_edus: - sent_edus_by_type.labels(edu.edu_type).inc() + self.sent_edus_by_type.labels(edu.edu_type).inc() except NotRetryingDestination as e: logger.debug( diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py index b7d1033351..602fd536f0 100644 --- a/synapse/handlers/appservice.py +++ b/synapse/handlers/appservice.py @@ -68,8 +68,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -events_processed_counter = Counter("synapse_handlers_appservice_events_processed", "") - class ApplicationServicesHandler: def __init__(self, hs: "HomeServer"): @@ -95,6 +93,12 @@ class ApplicationServicesHandler: name="appservice_ephemeral_events" ) + self.events_processed_counter = Counter( + "synapse_handlers_appservice_events_processed", + "", + registry=hs.metrics_collector_registry, + ) + def notify_interested_services(self, max_token: RoomStreamToken) -> None: """Notifies (pushes) all application services interested in this event. @@ -200,7 +204,7 @@ class ApplicationServicesHandler: "appservice_sender" ).set(upper_bound) - events_processed_counter.inc(len(events)) + self.events_processed_counter.inc(len(events)) event_processing_loop_room_count.labels("appservice_sender").inc( len(events_by_room) diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py index e96922c08d..d21b78f386 100644 --- a/synapse/handlers/auth.py +++ b/synapse/handlers/auth.py @@ -92,12 +92,6 @@ logger = logging.getLogger(__name__) INVALID_USERNAME_OR_PASSWORD = "Invalid username or password" -invalid_login_token_counter = Counter( - "synapse_user_login_invalid_login_tokens", - "Counts the number of rejected m.login.token on /login", - ["reason"], -) - def convert_client_dict_legacy_fields_to_identifier( submission: JsonDict, @@ -281,6 +275,13 @@ class AuthHandler: self.msc3861_oauth_delegation_enabled = hs.config.experimental.msc3861.enabled + self.invalid_login_token_counter = Counter( + "synapse_user_login_invalid_login_tokens", + "Counts the number of rejected m.login.token on /login", + ["reason"], + registry=hs.metrics_collector_registry, + ) + async def validate_user_via_ui_auth( self, requester: Requester, @@ -1477,11 +1478,11 @@ class AuthHandler: try: return await self.store.consume_login_token(login_token) except LoginTokenExpired: - invalid_login_token_counter.labels("expired").inc() + self.invalid_login_token_counter.labels("expired").inc() except LoginTokenReused: - invalid_login_token_counter.labels("reused").inc() + self.invalid_login_token_counter.labels("reused").inc() except NotFoundError: - invalid_login_token_counter.labels("not found").inc() + self.invalid_login_token_counter.labels("not found").inc() raise AuthError(403, "Invalid login token", errcode=Codes.FORBIDDEN) diff --git a/synapse/handlers/federation_event.py b/synapse/handlers/federation_event.py index 1e738f484f..42ef5f42f2 100644 --- a/synapse/handlers/federation_event.py +++ b/synapse/handlers/federation_event.py @@ -105,41 +105,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -soft_failed_event_counter = Counter( - "synapse_federation_soft_failed_events_total", - "Events received over federation that we marked as soft_failed", -) - -# Added to debug performance and track progress on optimizations -backfill_processing_after_timer = Histogram( - "synapse_federation_backfill_processing_after_time_seconds", - "sec", - [], - buckets=( - 0.1, - 0.25, - 0.5, - 1.0, - 2.5, - 5.0, - 7.5, - 10.0, - 15.0, - 20.0, - 25.0, - 30.0, - 40.0, - 50.0, - 60.0, - 80.0, - 100.0, - 120.0, - 150.0, - 180.0, - "+Inf", - ), -) - class FederationEventHandler: """Handles events that originated from federation. @@ -195,6 +160,43 @@ class FederationEventHandler: self._room_pdu_linearizer = Linearizer("fed_room_pdu") + self.soft_failed_event_counter = Counter( + "synapse_federation_soft_failed_events_total", + "Events received over federation that we marked as soft_failed", + registry=hs.metrics_collector_registry, + ) + + # Added to debug performance and track progress on optimizations + self.backfill_processing_after_timer = Histogram( + "synapse_federation_backfill_processing_after_time_seconds", + "sec", + [], + buckets=( + 0.1, + 0.25, + 0.5, + 1.0, + 2.5, + 5.0, + 7.5, + 10.0, + 15.0, + 20.0, + 25.0, + 30.0, + 40.0, + 50.0, + 60.0, + 80.0, + 100.0, + 120.0, + 150.0, + 180.0, + "+Inf", + ), + registry=hs.metrics_collector_registry, + ) + async def on_receive_pdu(self, origin: str, pdu: EventBase) -> None: """Process a PDU received via a federation /send/ transaction @@ -698,7 +700,7 @@ class FederationEventHandler: if not events: return - with backfill_processing_after_timer.time(): + with self.backfill_processing_after_timer.time(): # if there are any events in the wrong room, the remote server is buggy and # should not be trusted. for ev in events: @@ -2062,7 +2064,7 @@ class FederationEventHandler: "hs": origin, }, ) - soft_failed_event_counter.inc() + self.soft_failed_event_counter.inc() event.internal_metadata.soft_failed = True async def _load_or_fetch_auth_events_for_event( diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 390cafa8f6..bf0d98c7ea 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -137,26 +137,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -notified_presence_counter = Counter("synapse_handler_presence_notified_presence", "") -federation_presence_out_counter = Counter( - "synapse_handler_presence_federation_presence_out", "" -) -presence_updates_counter = Counter("synapse_handler_presence_presence_updates", "") -timers_fired_counter = Counter("synapse_handler_presence_timers_fired", "") -federation_presence_counter = Counter( - "synapse_handler_presence_federation_presence", "" -) -bump_active_time_counter = Counter("synapse_handler_presence_bump_active_time", "") - -get_updates_counter = Counter("synapse_handler_presence_get_updates", "", ["type"]) - -notify_reason_counter = Counter( - "synapse_handler_presence_notify_reason", "", ["locality", "reason"] -) -state_transition_counter = Counter( - "synapse_handler_presence_state_transition", "", ["locality", "from", "to"] -) - # If a user was last active in the last LAST_ACTIVE_GRANULARITY, consider them # "currently_active" LAST_ACTIVE_GRANULARITY = 60 * 1000 @@ -222,6 +202,19 @@ class BasePresenceHandler(abc.ABC): # The combined status across all user devices. self.user_to_current_state = {state.user_id: state for state in active_presence} + self.notify_reason_counter = Counter( + "synapse_handler_presence_notify_reason", + "", + ["locality", "reason"], + registry=hs.metrics_collector_registry, + ) + self.state_transition_counter = Counter( + "synapse_handler_presence_state_transition", + "", + ["locality", "from", "to"], + registry=hs.metrics_collector_registry, + ) + @abc.abstractmethod async def user_syncing( self, @@ -666,7 +659,7 @@ class WorkerPresenceHandler(BasePresenceHandler): old_state = self.user_to_current_state.get(new_state.user_id) self.user_to_current_state[new_state.user_id] = new_state is_mine = self.is_mine_id(new_state.user_id) - if not old_state or should_notify(old_state, new_state, is_mine): + if not old_state or should_notify(self, old_state, new_state, is_mine): state_to_notify.append(new_state) stream_id = token @@ -761,6 +754,38 @@ class PresenceHandler(BasePresenceHandler): "", [], lambda: len(self.user_to_current_state), + registry=hs.metrics_collector_registry, + ) + + self.notified_presence_counter = Counter( + "synapse_handler_presence_notified_presence", + "", + registry=hs.metrics_collector_registry, + ) + self.federation_presence_out_counter = Counter( + "synapse_handler_presence_federation_presence_out", + "", + registry=hs.metrics_collector_registry, + ) + self.presence_updates_counter = Counter( + "synapse_handler_presence_presence_updates", + "", + registry=hs.metrics_collector_registry, + ) + self.timers_fired_counter = Counter( + "synapse_handler_presence_timers_fired", + "", + registry=hs.metrics_collector_registry, + ) + self.federation_presence_counter = Counter( + "synapse_handler_presence_federation_presence", + "", + registry=hs.metrics_collector_registry, + ) + self.bump_active_time_counter = Counter( + "synapse_handler_presence_bump_active_time", + "", + registry=hs.metrics_collector_registry, ) # The per-device presence state, maps user to devices to per-device presence state. @@ -966,6 +991,7 @@ class PresenceHandler(BasePresenceHandler): ) new_state, should_notify, should_ping = handle_update( + self, prev_state, new_state, is_mine=self.is_mine_id(user_id), @@ -988,10 +1014,10 @@ class PresenceHandler(BasePresenceHandler): # TODO: We should probably ensure there are no races hereafter - presence_updates_counter.inc(len(new_states)) + self.presence_updates_counter.inc(len(new_states)) if to_notify: - notified_presence_counter.inc(len(to_notify)) + self.notified_presence_counter.inc(len(to_notify)) await self._persist_and_notify(list(to_notify.values())) self.unpersisted_users_changes |= {s.user_id for s in new_states} @@ -1010,7 +1036,7 @@ class PresenceHandler(BasePresenceHandler): if user_id not in to_notify } if to_federation_ping: - federation_presence_out_counter.inc(len(to_federation_ping)) + self.federation_presence_out_counter.inc(len(to_federation_ping)) hosts_to_states = await get_interested_remotes( self.store, @@ -1060,7 +1086,7 @@ class PresenceHandler(BasePresenceHandler): for user_id in users_to_check ] - timers_fired_counter.inc(len(states)) + self.timers_fired_counter.inc(len(states)) # Set of user ID & device IDs which are currently syncing. syncing_user_devices = { @@ -1094,7 +1120,7 @@ class PresenceHandler(BasePresenceHandler): user_id = user.to_string() - bump_active_time_counter.inc() + self.bump_active_time_counter.inc() now = self.clock.time_msec() @@ -1346,7 +1372,7 @@ class PresenceHandler(BasePresenceHandler): updates.append(prev_state.copy_and_replace(**new_fields)) if updates: - federation_presence_counter.inc(len(updates)) + self.federation_presence_counter.inc(len(updates)) await self._update_states(updates) async def set_state( @@ -1655,7 +1681,10 @@ class PresenceHandler(BasePresenceHandler): def should_notify( - old_state: UserPresenceState, new_state: UserPresenceState, is_mine: bool + base_presence_handler: BasePresenceHandler, + old_state: UserPresenceState, + new_state: UserPresenceState, + is_mine: bool, ) -> bool: """Decides if a presence state change should be sent to interested parties.""" user_location = "remote" @@ -1666,19 +1695,25 @@ def should_notify( return False if old_state.status_msg != new_state.status_msg: - notify_reason_counter.labels(user_location, "status_msg_change").inc() + base_presence_handler.notify_reason_counter.labels( + user_location, "status_msg_change" + ).inc() return True if old_state.state != new_state.state: - notify_reason_counter.labels(user_location, "state_change").inc() - state_transition_counter.labels( + base_presence_handler.notify_reason_counter.labels( + user_location, "state_change" + ).inc() + base_presence_handler.state_transition_counter.labels( user_location, old_state.state, new_state.state ).inc() return True if old_state.state == PresenceState.ONLINE: if new_state.currently_active != old_state.currently_active: - notify_reason_counter.labels(user_location, "current_active_change").inc() + base_presence_handler.notify_reason_counter.labels( + user_location, "current_active_change" + ).inc() return True if ( @@ -1687,14 +1722,14 @@ def should_notify( ): # Only notify about last active bumps if we're not currently active if not new_state.currently_active: - notify_reason_counter.labels( + base_presence_handler.notify_reason_counter.labels( user_location, "last_active_change_online" ).inc() return True elif new_state.last_active_ts - old_state.last_active_ts > LAST_ACTIVE_GRANULARITY: # Always notify for a transition where last active gets bumped. - notify_reason_counter.labels( + base_presence_handler.notify_reason_counter.labels( user_location, "last_active_change_not_online" ).inc() return True @@ -1764,6 +1799,13 @@ class PresenceEventSource(EventSource[int, UserPresenceState]): self.clock = hs.get_clock() self.store = hs.get_datastores().main + self.get_updates_counter = Counter( + "synapse_handler_presence_get_updates", + "", + ["type"], + registry=hs.metrics_collector_registry, + ) + async def get_new_events( self, user: UserID, @@ -1870,7 +1912,7 @@ class PresenceEventSource(EventSource[int, UserPresenceState]): # If we have the full list of changes for presence we can # simply check which ones share a room with the user. - get_updates_counter.labels("stream").inc() + self.get_updates_counter.labels("stream").inc() sharing_users = await self.store.do_users_share_a_room( user_id, updated_users @@ -1883,7 +1925,7 @@ class PresenceEventSource(EventSource[int, UserPresenceState]): else: # Too many possible updates. Find all users we can see and check # if any of them have changed. - get_updates_counter.labels("full").inc() + self.get_updates_counter.labels("full").inc() users_interested_in = ( await self.store.get_users_who_share_room_with_user(user_id) @@ -2130,6 +2172,7 @@ def handle_timeout( def handle_update( + base_presence_handler: BasePresenceHandler, prev_state: UserPresenceState, new_state: UserPresenceState, is_mine: bool, @@ -2213,7 +2256,7 @@ def handle_update( ) # Check whether the change was something worth notifying about - if should_notify(prev_state, new_state, is_mine): + if should_notify(base_presence_handler, prev_state, new_state, is_mine): new_state = new_state.copy_and_replace(last_federation_update_ts=now) persist_and_notify = True