Merge branch 'develop' into application-services-txn-reliability

Conflicts:
	synapse/storage/appservice.py

1 files changed, 45 insertions, 0 deletions

diff --git a/synapse/notifier.py b/synapse/notifier.py
index df13e8ddb6..7121d659d0 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -19,12 +19,27 @@ from synapse.util.logutils import log_function
 from synapse.util.logcontext import PreserveLoggingContext
 from synapse.util.async import run_on_reactor
 from synapse.types import StreamToken
+import synapse.metrics
 
 import logging
 
 
 logger = logging.getLogger(__name__)
 
+metrics = synapse.metrics.get_metrics_for(__name__)
+
+notified_events_counter = metrics.register_counter("notified_events")
+
+
+# TODO(paul): Should be shared somewhere
+def count(func, l):
+    """Return the number of items in l for which func returns true."""
+    n = 0
+    for x in l:
+        if func(x):
+            n += 1
+    return n
+
 
 class _NotificationListener(object):
     """ This represents a single client connection to the events stream.
@@ -59,6 +74,7 @@ class _NotificationListener(object):
 
         try:
             self.deferred.callback(result)
+            notified_events_counter.inc_by(len(events))
         except defer.AlreadyCalledError:
             pass
 
@@ -95,6 +111,35 @@ class Notifier(object):
             "user_joined_room", self._user_joined_room
         )
 
+        # This is not a very cheap test to perform, but it's only executed
+        # when rendering the metrics page, which is likely once per minute at
+        # most when scraping it.
+        def count_listeners():
+            all_listeners = set()
+
+            for x in self.room_to_listeners.values():
+                all_listeners |= x
+            for x in self.user_to_listeners.values():
+                all_listeners |= x
+            for x in self.appservice_to_listeners.values():
+                all_listeners |= x
+
+            return len(all_listeners)
+        metrics.register_callback("listeners", count_listeners)
+
+        metrics.register_callback(
+            "rooms",
+            lambda: count(bool, self.room_to_listeners.values()),
+        )
+        metrics.register_callback(
+            "users",
+            lambda: count(bool, self.user_to_listeners.values()),
+        )
+        metrics.register_callback(
+            "appservices",
+            lambda: count(bool, self.appservice_to_listeners.values()),
+        )
+
     @log_function
     @defer.inlineCallbacks
     def on_new_room_event(self, event, extra_users=[]):