From df9f72d9e5fe264b86005208e0f096156eb03e4b Mon Sep 17 00:00:00 2001
From: Amber Brown <hawkowl@atleastfornow.net>
Date: Mon, 21 May 2018 19:47:37 -0500
Subject: replacing portions

---
 synapse/replication/tcp/protocol.py | 88 ++++++++++++++-----------------------
 1 file changed, 34 insertions(+), 54 deletions(-)

(limited to 'synapse/replication/tcp')

diff --git a/synapse/replication/tcp/protocol.py b/synapse/replication/tcp/protocol.py
index d7d38464b2..5848f57c5e 100644
--- a/synapse/replication/tcp/protocol.py
+++ b/synapse/replication/tcp/protocol.py
@@ -60,19 +60,19 @@ from .commands import (
 )
 from .streams import STREAMS_MAP
 
+from synapse.metrics import LaterGauge
 from synapse.util.stringutils import random_string
-from synapse.metrics.metric import CounterMetric
+
+from prometheus_client import Counter
+
+from collections import defaultdict
 
 import logging
-import synapse.metrics
 import struct
 import fcntl
 
-
-metrics = synapse.metrics.get_metrics_for(__name__)
-
-connection_close_counter = metrics.register_counter(
-    "close_reason", labels=["reason_type"],
+connection_close_counter = Counter(
+    "synapse_replication_tcp_protocol_close_reason", "", ["reason_type"],
 )
 
 
@@ -136,12 +136,8 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
         # The LoopingCall for sending pings.
         self._send_ping_loop = None
 
-        self.inbound_commands_counter = CounterMetric(
-            "inbound_commands", labels=["command"],
-        )
-        self.outbound_commands_counter = CounterMetric(
-            "outbound_commands", labels=["command"],
-        )
+        self.inbound_commands_counter = defaultdict(int)
+        self.outbound_commands_counter = defaultdict(int)
 
     def connectionMade(self):
         logger.info("[%s] Connection established", self.id())
@@ -201,7 +197,7 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
 
         self.last_received_command = self.clock.time_msec()
 
-        self.inbound_commands_counter.inc(cmd_name)
+        self.inbound_commands_counter[cmd_name] = self.inbound_commands_counter[cmd_name] + 1
 
         cmd_cls = COMMAND_MAP[cmd_name]
         try:
@@ -251,8 +247,7 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
             self._queue_command(cmd)
             return
 
-        self.outbound_commands_counter.inc(cmd.NAME)
-
+        self.outbound_commands_counter[cmd.NAME] = self.outbound_commands_counter[cmd.NAME] + 1
         string = "%s %s" % (cmd.NAME, cmd.to_line(),)
         if "\n" in string:
             raise Exception("Unexpected newline in command: %r", string)
@@ -317,9 +312,9 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
     def connectionLost(self, reason):
         logger.info("[%s] Replication connection closed: %r", self.id(), reason)
         if isinstance(reason, Failure):
-            connection_close_counter.inc(reason.type.__name__)
+            connection_close_counter.labels(reason.type.__name__).inc()
         else:
-            connection_close_counter.inc(reason.__class__.__name__)
+            connection_close_counter.labels(reason.__class__.__name__).inc()
 
         try:
             # Remove us from list of connections to be monitored
@@ -566,14 +561,12 @@ class ClientReplicationStreamProtocol(BaseReplicationStreamProtocol):
 
 # The following simply registers metrics for the replication connections
 
-metrics.register_callback(
-    "pending_commands",
+pending_commands = LaterGauge(
+    "pending_commands", "", ["name", "conn_id"],
     lambda: {
         (p.name, p.conn_id): len(p.pending_commands)
         for p in connected_connections
-    },
-    labels=["name", "conn_id"],
-)
+    })
 
 
 def transport_buffer_size(protocol):
@@ -583,14 +576,12 @@ def transport_buffer_size(protocol):
     return 0
 
 
-metrics.register_callback(
-    "transport_send_buffer",
+transport_send_buffer = LaterGauge(
+    "synapse_replication_tcp_transport_send_buffer", "", ["name", "conn_id"],
     lambda: {
         (p.name, p.conn_id): transport_buffer_size(p)
         for p in connected_connections
-    },
-    labels=["name", "conn_id"],
-)
+    })
 
 
 def transport_kernel_read_buffer_size(protocol, read=True):
@@ -608,48 +599,37 @@ def transport_kernel_read_buffer_size(protocol, read=True):
     return 0
 
 
-metrics.register_callback(
-    "transport_kernel_send_buffer",
+tcp_transport_kernel_send_buffer = LaterGauge(
+    "synapse_replication_tcp_transport_kernel_send_buffer", "", ["name", "conn_id"],
     lambda: {
         (p.name, p.conn_id): transport_kernel_read_buffer_size(p, False)
         for p in connected_connections
-    },
-    labels=["name", "conn_id"],
-)
+    })
 
 
-metrics.register_callback(
-    "transport_kernel_read_buffer",
+tcp_transport_kernel_read_buffer = LaterGauge(
+    "synapse_replication_tcp_transport_kernel_read_buffer", "", ["name", "conn_id"],
     lambda: {
         (p.name, p.conn_id): transport_kernel_read_buffer_size(p, True)
         for p in connected_connections
-    },
-    labels=["name", "conn_id"],
-)
+    })
 
 
-metrics.register_callback(
-    "inbound_commands",
+tcp_inbound_commands = LaterGauge(
+    "synapse_replication_tcp_inbound_commands", "", ["command", "name", "conn_id"],
     lambda: {
         (k[0], p.name, p.conn_id): count
         for p in connected_connections
-        for k, count in p.inbound_commands_counter.counts.iteritems()
-    },
-    labels=["command", "name", "conn_id"],
-)
+        for k, count in p.inbound_commands_counter.items()
+    })
 
-metrics.register_callback(
-    "outbound_commands",
+tcp_outbound_commands = LaterGauge(
+    "synapse_replication_tcp_outbound_commands", "", ["command", "name", "conn_id"],
     lambda: {
         (k[0], p.name, p.conn_id): count
         for p in connected_connections
-        for k, count in p.outbound_commands_counter.counts.iteritems()
-    },
-    labels=["command", "name", "conn_id"],
-)
+        for k, count in p.outbound_commands_counter.items()
+    })
 
 # number of updates received for each RDATA stream
-inbound_rdata_count = metrics.register_counter(
-    "inbound_rdata_count",
-    labels=["stream_name"],
-)
+inbound_rdata_count = Counter("synapse_replication_tcp_inbound_rdata_count", "", ["stream_name"])
-- 
cgit 1.4.1


From fcc525b0b705703fead3d703c0df62a264ff8ce8 Mon Sep 17 00:00:00 2001
From: Amber Brown <hawkowl@atleastfornow.net>
Date: Mon, 21 May 2018 19:48:57 -0500
Subject: rest of the changes

---
 synapse/http/request_metrics.py          | 106 ++++-------------
 synapse/http/server.py                   |   4 +-
 synapse/push/bulk_push_rule_evaluator.py |  28 ++---
 synapse/replication/tcp/resource.py      |  30 +++--
 synapse/storage/_base.py                 |  17 ++-
 synapse/storage/events.py                |  28 ++---
 tests/metrics/__init__.py                |   0
 tests/metrics/test_metric.py             | 192 -------------------------------
 8 files changed, 68 insertions(+), 337 deletions(-)
 delete mode 100644 tests/metrics/__init__.py
 delete mode 100644 tests/metrics/test_metric.py

(limited to 'synapse/replication/tcp')

diff --git a/synapse/http/request_metrics.py b/synapse/http/request_metrics.py
index 8c850bf23f..34a730d5bc 100644
--- a/synapse/http/request_metrics.py
+++ b/synapse/http/request_metrics.py
@@ -16,86 +16,38 @@
 
 import logging
 
-import synapse.metrics
+from prometheus_client.core import Counter, Histogram
+
 from synapse.util.logcontext import LoggingContext
 
 logger = logging.getLogger(__name__)
 
-metrics = synapse.metrics.get_metrics_for("synapse.http.server")
 
 # total number of responses served, split by method/servlet/tag
-response_count = metrics.register_counter(
-    "response_count",
-    labels=["method", "servlet", "tag"],
-    alternative_names=(
-        # the following are all deprecated aliases for the same metric
-        metrics.name_prefix + x for x in (
-            "_requests",
-            "_response_time:count",
-            "_response_ru_utime:count",
-            "_response_ru_stime:count",
-            "_response_db_txn_count:count",
-            "_response_db_txn_duration:count",
-        )
-    )
-)
+response_count = Counter("synapse_http_server_response_count", "", ["method", "servlet", "tag"])
 
-requests_counter = metrics.register_counter(
-    "requests_received",
-    labels=["method", "servlet", ],
-)
+requests_counter = Counter("synapse_http_server_requests_received", "", ["method", "servlet"])
 
-outgoing_responses_counter = metrics.register_counter(
-    "responses",
-    labels=["method", "code"],
-)
+outgoing_responses_counter = Counter("synapse_http_server_responses", "", ["method", "code"])
 
-response_timer = metrics.register_counter(
-    "response_time_seconds",
-    labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_time:total",
-    ),
-)
+response_timer = Histogram("synapse_http_server_response_time_seconds", "", ["method", "servlet", "tag"])
 
-response_ru_utime = metrics.register_counter(
-    "response_ru_utime_seconds", labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_ru_utime:total",
-    ),
-)
+response_ru_utime = Counter("synapse_http_server_response_ru_utime_seconds", "", ["method", "servlet", "tag"])
 
-response_ru_stime = metrics.register_counter(
-    "response_ru_stime_seconds", labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_ru_stime:total",
-    ),
-)
+response_ru_stime = Counter("synapse_http_server_response_ru_stime_seconds", "", ["method", "servlet", "tag"])
 
-response_db_txn_count = metrics.register_counter(
-    "response_db_txn_count", labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_db_txn_count:total",
-    ),
-)
+response_db_txn_count = Counter("synapse_http_server_response_db_txn_count", "", ["method", "servlet", "tag"])
 
 # seconds spent waiting for db txns, excluding scheduling time, when processing
 # this request
-response_db_txn_duration = metrics.register_counter(
-    "response_db_txn_duration_seconds", labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_db_txn_duration:total",
-    ),
-)
+response_db_txn_duration = Counter("synapse_http_server_response_db_txn_duration_seconds", "", ["method", "servlet", "tag"])
 
 # seconds spent waiting for a db connection, when processing this request
-response_db_sched_duration = metrics.register_counter(
-    "response_db_sched_duration_seconds", labels=["method", "servlet", "tag"]
+response_db_sched_duration = Counter("synapse_http_request_response_db_sched_duration_seconds", "", ["method", "servlet", "tag"]
 )
 
 # size in bytes of the response written
-response_size = metrics.register_counter(
-    "response_size", labels=["method", "servlet", "tag"]
+response_size = Counter("synapse_http_request_response_size", "", ["method", "servlet", "tag"]
 )
 
 
@@ -119,31 +71,19 @@ class RequestMetrics(object):
                 )
                 return
 
-        outgoing_responses_counter.inc(request.method, str(request.code))
+        outgoing_responses_counter.labels(request.method, str(request.code)).inc()
 
-        response_count.inc(request.method, self.name, tag)
+        response_count.labels(request.method, self.name, tag).inc()
 
-        response_timer.inc_by(
-            time_msec - self.start, request.method,
-            self.name, tag
-        )
+        response_timer.labels(request.method, self.name, tag).observe(time_msec - self.start)
 
         ru_utime, ru_stime = context.get_resource_usage()
 
-        response_ru_utime.inc_by(
-            ru_utime, request.method, self.name, tag
-        )
-        response_ru_stime.inc_by(
-            ru_stime, request.method, self.name, tag
-        )
-        response_db_txn_count.inc_by(
-            context.db_txn_count, request.method, self.name, tag
-        )
-        response_db_txn_duration.inc_by(
-            context.db_txn_duration_ms / 1000., request.method, self.name, tag
-        )
-        response_db_sched_duration.inc_by(
-            context.db_sched_duration_ms / 1000., request.method, self.name, tag
-        )
-
-        response_size.inc_by(request.sentLength, request.method, self.name, tag)
+        response_ru_utime.labels(request.method, self.name, tag).inc(ru_utime)
+        response_ru_stime.labels(request.method, self.name, tag).inc(ru_stime)
+        response_db_txn_count.labels(request.method, self.name, tag).inc(context.db_txn_count)
+        response_db_txn_duration.labels(request.method, self.name, tag).inc(context.db_txn_duration_ms / 1000.)
+        response_db_sched_duration.labels(request.method, self.name, tag).inc(
+            context.db_sched_duration_ms / 1000.)
+
+        response_size.labels(request.method, self.name, tag).inc(request.sentLength)
diff --git a/synapse/http/server.py b/synapse/http/server.py
index b6e2ae14a2..f72d986288 100644
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -138,8 +138,8 @@ def wrap_request_handler_with_logging(h):
                         # dispatching to the handler, so that the handler
                         # can update the servlet name in the request
                         # metrics
-                        requests_counter.inc(request.method,
-                                             request.request_metrics.name)
+                        requests_counter.labels(request.method,
+                                                request.request_metrics.name).inc()
                         yield d
     return wrapped_request_handler
 
diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py
index 7c680659b6..6fcca5e260 100644
--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -22,35 +22,29 @@ from .push_rule_evaluator import PushRuleEvaluatorForEvent
 
 from synapse.event_auth import get_user_power_level
 from synapse.api.constants import EventTypes, Membership
-from synapse.metrics import get_metrics_for
-from synapse.util.caches import metrics as cache_metrics
+from synapse.util.caches import register_cache
 from synapse.util.caches.descriptors import cached
 from synapse.util.async import Linearizer
 from synapse.state import POWER_KEY
 
 from collections import namedtuple
-
+from prometheus_client import Counter
 
 logger = logging.getLogger(__name__)
 
 
 rules_by_room = {}
 
-push_metrics = get_metrics_for(__name__)
 
-push_rules_invalidation_counter = push_metrics.register_counter(
-    "push_rules_invalidation_counter"
-)
-push_rules_state_size_counter = push_metrics.register_counter(
-    "push_rules_state_size_counter"
-)
+push_rules_invalidation_counter = Counter("synapse_push_bulk_push_role_evaluator_push_rules_invalidation_counter", "")
+push_rules_state_size_counter = Counter("synapse_push_bulk_push_role_evaluator_push_rules_state_size_counter", "")
 
 # Measures whether we use the fast path of using state deltas, or if we have to
 # recalculate from scratch
-push_rules_delta_state_cache_metric = cache_metrics.register_cache(
+push_rules_delta_state_cache_metric = register_cache(
     "cache",
-    size_callback=lambda: 0,  # Meaningless size, as this isn't a cache that stores values
-    cache_name="push_rules_delta_state_cache_metric",
+    "push_rules_delta_state_cache_metric",
+    cache=[],  # Meaningless size, as this isn't a cache that stores values
 )
 
 
@@ -64,10 +58,10 @@ class BulkPushRuleEvaluator(object):
         self.store = hs.get_datastore()
         self.auth = hs.get_auth()
 
-        self.room_push_rule_cache_metrics = cache_metrics.register_cache(
+        self.room_push_rule_cache_metrics = register_cache(
             "cache",
-            size_callback=lambda: 0,  # There's not good value for this
-            cache_name="room_push_rule_cache",
+            "room_push_rule_cache",
+            cache=[],  # Meaningless size, as this isn't a cache that stores values
         )
 
     @defer.inlineCallbacks
@@ -309,7 +303,7 @@ class RulesForRoom(object):
                 current_state_ids = context.current_state_ids
                 push_rules_delta_state_cache_metric.inc_misses()
 
-            push_rules_state_size_counter.inc_by(len(current_state_ids))
+            push_rules_state_size_counter.inc(len(current_state_ids))
 
             logger.debug(
                 "Looking for member changes in %r %r", state_group, current_state_ids
diff --git a/synapse/replication/tcp/resource.py b/synapse/replication/tcp/resource.py
index a41af4fd6c..0e6b1957c6 100644
--- a/synapse/replication/tcp/resource.py
+++ b/synapse/replication/tcp/resource.py
@@ -22,20 +22,19 @@ from .streams import STREAMS_MAP, FederationStream
 from .protocol import ServerReplicationStreamProtocol
 
 from synapse.util.metrics import Measure, measure_func
+from synapse.metrics import LaterGauge
 
 import logging
-import synapse.metrics
 
+from prometheus_client import Counter
 
-metrics = synapse.metrics.get_metrics_for(__name__)
-stream_updates_counter = metrics.register_counter(
-    "stream_updates", labels=["stream_name"]
+stream_updates_counter = Counter("synapse_replication_tcp_resource_stream_updates", "", ["stream_name"]
 )
-user_sync_counter = metrics.register_counter("user_sync")
-federation_ack_counter = metrics.register_counter("federation_ack")
-remove_pusher_counter = metrics.register_counter("remove_pusher")
-invalidate_cache_counter = metrics.register_counter("invalidate_cache")
-user_ip_cache_counter = metrics.register_counter("user_ip_cache")
+user_sync_counter = Counter("synapse_replication_tcp_resource_user_sync", "")
+federation_ack_counter = Counter("synapse_replication_tcp_resource_federation_ack", "")
+remove_pusher_counter = Counter("synapse_replication_tcp_resource_remove_pusher", "")
+invalidate_cache_counter = Counter("synapse_replication_tcp_resource_invalidate_cache", "")
+user_ip_cache_counter = Counter("synapse_replication_tcp_resource_user_ip_cache", "")
 
 logger = logging.getLogger(__name__)
 
@@ -73,7 +72,8 @@ class ReplicationStreamer(object):
         # Current connections.
         self.connections = []
 
-        metrics.register_callback("total_connections", lambda: len(self.connections))
+        l = LaterGauge("synapse_replication_tcp_resource_total_connections", "", [], lambda: len(self.connections))
+        l.register()
 
         # List of streams that clients can subscribe to.
         # We only support federation stream if federation sending hase been
@@ -85,17 +85,15 @@ class ReplicationStreamer(object):
 
         self.streams_by_name = {stream.NAME: stream for stream in self.streams}
 
-        metrics.register_callback(
-            "connections_per_stream",
+        LaterGauge(
+            "synapse_replication_tcp_resource_connections_per_stream", "", ["stream_name"],
             lambda: {
                 (stream_name,): len([
                     conn for conn in self.connections
                     if stream_name in conn.replication_streams
                 ])
                 for stream_name in self.streams_by_name
-            },
-            labels=["stream_name"],
-        )
+            }).register()
 
         self.federation_sender = None
         if not hs.config.send_federation:
@@ -175,7 +173,7 @@ class ReplicationStreamer(object):
                             logger.info(
                                 "Streaming: %s -> %s", stream.NAME, updates[-1][0]
                             )
-                            stream_updates_counter.inc_by(len(updates), stream.NAME)
+                            stream_updates_counter.labels(stream.NAME).inc(len(updates))
 
                         # Some streams return multiple rows with the same stream IDs,
                         # we need to make sure they get sent out in batches. We do
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 2262776ab2..d1b625dc30 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -18,8 +18,8 @@ from synapse.api.errors import StoreError
 from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
 from synapse.util.caches.descriptors import Cache
 from synapse.storage.engines import PostgresEngine
-import synapse.metrics
 
+from prometheus_client import Histogram
 
 from twisted.internet import defer
 
@@ -34,13 +34,10 @@ sql_logger = logging.getLogger("synapse.storage.SQL")
 transaction_logger = logging.getLogger("synapse.storage.txn")
 perf_logger = logging.getLogger("synapse.storage.TIME")
 
+sql_scheduling_timer = Histogram("synapse_storage_schedule_time", "")
 
-metrics = synapse.metrics.get_metrics_for("synapse.storage")
-
-sql_scheduling_timer = metrics.register_distribution("schedule_time")
-
-sql_query_timer = metrics.register_distribution("query_time", labels=["verb"])
-sql_txn_timer = metrics.register_distribution("transaction_time", labels=["desc"])
+sql_query_timer = Histogram("synapse_storage_query_time", "", ["verb"])
+sql_txn_timer = Histogram("synapse_storage_transaction_time", "", ["desc"])
 
 
 class LoggingTransaction(object):
@@ -117,7 +114,7 @@ class LoggingTransaction(object):
         finally:
             msecs = (time.time() * 1000) - start
             sql_logger.debug("[SQL time] {%s} %f", self.name, msecs)
-            sql_query_timer.inc_by(msecs, sql.split()[0])
+            sql_query_timer.labels(sql.split()[0]).observe(msecs)
 
 
 class PerformanceCounters(object):
@@ -287,7 +284,7 @@ class SQLBaseStore(object):
 
             self._current_txn_total_time += duration
             self._txn_perf_counters.update(desc, start, end)
-            sql_txn_timer.inc_by(duration, desc)
+            sql_txn_timer.labels(desc).observe(duration)
 
     @defer.inlineCallbacks
     def runInteraction(self, desc, func, *args, **kwargs):
@@ -349,7 +346,7 @@ class SQLBaseStore(object):
         def inner_func(conn, *args, **kwargs):
             with LoggingContext("runWithConnection") as context:
                 sched_duration_ms = time.time() * 1000 - start_time
-                sql_scheduling_timer.inc_by(sched_duration_ms)
+                sql_scheduling_timer.observe(sched_duration_ms)
                 current_context.add_database_scheduled(sched_duration_ms)
 
                 if self.database_engine.is_connection_closed(conn):
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 05cde96afc..96b48cfdbb 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -40,30 +40,24 @@ import synapse.metrics
 from synapse.events import EventBase    # noqa: F401
 from synapse.events.snapshot import EventContext   # noqa: F401
 
-logger = logging.getLogger(__name__)
+from prometheus_client import Counter
 
+logger = logging.getLogger(__name__)
 
-metrics = synapse.metrics.get_metrics_for(__name__)
-persist_event_counter = metrics.register_counter("persisted_events")
-event_counter = metrics.register_counter(
-    "persisted_events_sep", labels=["type", "origin_type", "origin_entity"]
-)
+persist_event_counter = Counter("synapse_storage_events_persisted_events", "")
+event_counter = Counter("synapse_storage_events_persisted_events_sep", "", ["type", "origin_type", "origin_entity"])
 
 # The number of times we are recalculating the current state
-state_delta_counter = metrics.register_counter(
-    "state_delta",
-)
+state_delta_counter = Counter("synapse_storage_events_state_delta", "")
+
 # The number of times we are recalculating state when there is only a
 # single forward extremity
-state_delta_single_event_counter = metrics.register_counter(
-    "state_delta_single_event",
-)
+state_delta_single_event_counter = Counter("synapse_storage_events_state_delta_single_event", "")
+
 # The number of times we are reculating state when we could have resonably
 # calculated the delta when we calculated the state for an event we were
 # persisting.
-state_delta_reuse_delta_counter = metrics.register_counter(
-    "state_delta_reuse_delta",
-)
+state_delta_reuse_delta_counter = Counter("synapse_storage_events_state_delta_reuse_delta", "")
 
 
 def encode_json(json_object):
@@ -445,7 +439,7 @@ class EventsStore(EventsWorkerStore):
                     state_delta_for_room=state_delta_for_room,
                     new_forward_extremeties=new_forward_extremeties,
                 )
-                persist_event_counter.inc_by(len(chunk))
+                persist_event_counter.inc(len(chunk))
                 synapse.metrics.event_persisted_position.set(
                     chunk[-1][0].internal_metadata.stream_ordering,
                 )
@@ -460,7 +454,7 @@ class EventsStore(EventsWorkerStore):
                         origin_type = "remote"
                         origin_entity = get_domain_from_id(event.sender)
 
-                    event_counter.inc(event.type, origin_type, origin_entity)
+                    event_counter.labels(event.type, origin_type, origin_entity).inc()
 
                 for room_id, new_state in current_state_for_room.iteritems():
                     self.get_current_state_ids.prefill(
diff --git a/tests/metrics/__init__.py b/tests/metrics/__init__.py
deleted file mode 100644
index e69de29bb2..0000000000
diff --git a/tests/metrics/test_metric.py b/tests/metrics/test_metric.py
deleted file mode 100644
index 069c0be762..0000000000
--- a/tests/metrics/test_metric.py
+++ /dev/null
@@ -1,192 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2015, 2016 OpenMarket Ltd
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-from tests import unittest
-
-from synapse.metrics.metric import (
-    CounterMetric, CallbackMetric, DistributionMetric, CacheMetric,
-    _escape_label_value,
-)
-
-
-class CounterMetricTestCase(unittest.TestCase):
-
-    def test_scalar(self):
-        counter = CounterMetric("scalar")
-
-        self.assertEquals(counter.render(), [
-            'scalar 0',
-        ])
-
-        counter.inc()
-
-        self.assertEquals(counter.render(), [
-            'scalar 1',
-        ])
-
-        counter.inc_by(2)
-
-        self.assertEquals(counter.render(), [
-            'scalar 3'
-        ])
-
-    def test_vector(self):
-        counter = CounterMetric("vector", labels=["method"])
-
-        # Empty counter doesn't yet know what values it has
-        self.assertEquals(counter.render(), [])
-
-        counter.inc("GET")
-
-        self.assertEquals(counter.render(), [
-            'vector{method="GET"} 1',
-        ])
-
-        counter.inc("GET")
-        counter.inc("PUT")
-
-        self.assertEquals(counter.render(), [
-            'vector{method="GET"} 2',
-            'vector{method="PUT"} 1',
-        ])
-
-
-class CallbackMetricTestCase(unittest.TestCase):
-
-    def test_scalar(self):
-        d = dict()
-
-        metric = CallbackMetric("size", lambda: len(d))
-
-        self.assertEquals(metric.render(), [
-            'size 0',
-        ])
-
-        d["key"] = "value"
-
-        self.assertEquals(metric.render(), [
-            'size 1',
-        ])
-
-    def test_vector(self):
-        vals = dict()
-
-        metric = CallbackMetric("values", lambda: vals, labels=["type"])
-
-        self.assertEquals(metric.render(), [])
-
-        # Keys have to be tuples, even if they're 1-element
-        vals[("foo",)] = 1
-        vals[("bar",)] = 2
-
-        self.assertEquals(metric.render(), [
-            'values{type="bar"} 2',
-            'values{type="foo"} 1',
-        ])
-
-
-class DistributionMetricTestCase(unittest.TestCase):
-
-    def test_scalar(self):
-        metric = DistributionMetric("thing")
-
-        self.assertEquals(metric.render(), [
-            'thing:count 0',
-            'thing:total 0',
-        ])
-
-        metric.inc_by(500)
-
-        self.assertEquals(metric.render(), [
-            'thing:count 1',
-            'thing:total 500',
-        ])
-
-    def test_vector(self):
-        metric = DistributionMetric("queries", labels=["verb"])
-
-        self.assertEquals(metric.render(), [])
-
-        metric.inc_by(300, "SELECT")
-        metric.inc_by(200, "SELECT")
-        metric.inc_by(800, "INSERT")
-
-        self.assertEquals(metric.render(), [
-            'queries:count{verb="INSERT"} 1',
-            'queries:count{verb="SELECT"} 2',
-            'queries:total{verb="INSERT"} 800',
-            'queries:total{verb="SELECT"} 500',
-        ])
-
-
-class CacheMetricTestCase(unittest.TestCase):
-
-    def test_cache(self):
-        d = dict()
-
-        metric = CacheMetric("cache", lambda: len(d), "cache_name")
-
-        self.assertEquals(metric.render(), [
-            'cache:hits{name="cache_name"} 0',
-            'cache:total{name="cache_name"} 0',
-            'cache:size{name="cache_name"} 0',
-            'cache:evicted_size{name="cache_name"} 0',
-        ])
-
-        metric.inc_misses()
-        d["key"] = "value"
-
-        self.assertEquals(metric.render(), [
-            'cache:hits{name="cache_name"} 0',
-            'cache:total{name="cache_name"} 1',
-            'cache:size{name="cache_name"} 1',
-            'cache:evicted_size{name="cache_name"} 0',
-        ])
-
-        metric.inc_hits()
-
-        self.assertEquals(metric.render(), [
-            'cache:hits{name="cache_name"} 1',
-            'cache:total{name="cache_name"} 2',
-            'cache:size{name="cache_name"} 1',
-            'cache:evicted_size{name="cache_name"} 0',
-        ])
-
-        metric.inc_evictions(2)
-
-        self.assertEquals(metric.render(), [
-            'cache:hits{name="cache_name"} 1',
-            'cache:total{name="cache_name"} 2',
-            'cache:size{name="cache_name"} 1',
-            'cache:evicted_size{name="cache_name"} 2',
-        ])
-
-
-class LabelValueEscapeTestCase(unittest.TestCase):
-    def test_simple(self):
-        string = "safjhsdlifhyskljfksdfh"
-        self.assertEqual(string, _escape_label_value(string))
-
-    def test_escape(self):
-        self.assertEqual(
-            "abc\\\"def\\nghi\\\\",
-            _escape_label_value("abc\"def\nghi\\"),
-        )
-
-    def test_sequence_of_escapes(self):
-        self.assertEqual(
-            "abc\\\"def\\nghi\\\\\\n",
-            _escape_label_value("abc\"def\nghi\\\n"),
-        )
-- 
cgit 1.4.1


From 8f5a688d420c8f6b51826c561da9094b76fbea1e Mon Sep 17 00:00:00 2001
From: Amber Brown <hawkowl@atleastfornow.net>
Date: Tue, 22 May 2018 10:56:03 -0500
Subject: cleanups, self-registration

---
 synapse/handlers/presence.py        |  7 ++++---
 synapse/http/request_metrics.py     | 15 +++------------
 synapse/replication/tcp/resource.py |  9 +++++----
 3 files changed, 12 insertions(+), 19 deletions(-)

(limited to 'synapse/replication/tcp')

diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index aca822c46a..4ee87d5714 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -142,8 +142,9 @@ class PresenceHandler(object):
         }
 
         LaterGauge(
-            "user_to_current_state_size", "", [], lambda: len(self.user_to_current_state)
-        ).register()
+            "synapse_handlers_presence_user_to_current_state_size", "", [],
+            lambda: len(self.user_to_current_state)
+        )
 
         now = self.clock.time_msec()
         for state in active_presence:
@@ -212,7 +213,7 @@ class PresenceHandler(object):
             60 * 1000,
         )
 
-        LaterGauge("wheel_timer_size", "", [], lambda: len(self.wheel_timer)).register()
+        LaterGauge("synapse_handlers_presence_wheel_timer_size", "", [], lambda: len(self.wheel_timer))
 
     @defer.inlineCallbacks
     def _on_shutdown(self):
diff --git a/synapse/http/request_metrics.py b/synapse/http/request_metrics.py
index 0984870e7e..e7f1bfc4ae 100644
--- a/synapse/http/request_metrics.py
+++ b/synapse/http/request_metrics.py
@@ -71,24 +71,14 @@ in_flight_requests_db_sched_duration = Counter("synapse_http_request_in_flight_r
 _in_flight_requests = set()
 
 
-def _collect_in_flight():
-    """Called just before metrics are collected, so we use it to update all
-    the in flight request metrics
-    """
-
-    for rm in _in_flight_requests:
-        rm.update_metrics()
-
-
-metrics.register_collector(_collect_in_flight)
-
-
 def _get_in_flight_counts():
     """Returns a count of all in flight requests by (method, server_name)
 
     Returns:
         dict[tuple[str, str], int]
     """
+    for rm in _in_flight_requests:
+        rm.update_metrics()
 
     # Map from (method, name) -> int, the number of in flight requests of that
     # type
@@ -99,6 +89,7 @@ def _get_in_flight_counts():
 
     return counts
 
+
 LaterGauge(
     "synapse_http_request_metrics_in_flight_requests_count", "",
     ["method", "servlet"],
diff --git a/synapse/replication/tcp/resource.py b/synapse/replication/tcp/resource.py
index ed251c79ac..1ca30bc31a 100644
--- a/synapse/replication/tcp/resource.py
+++ b/synapse/replication/tcp/resource.py
@@ -73,8 +73,8 @@ class ReplicationStreamer(object):
         # Current connections.
         self.connections = []
 
-        l = LaterGauge("synapse_replication_tcp_resource_total_connections", "", [], lambda: len(self.connections))
-        l.register()
+        LaterGauge("synapse_replication_tcp_resource_total_connections", "", [],
+                   lambda: len(self.connections))
 
         # List of streams that clients can subscribe to.
         # We only support federation stream if federation sending hase been
@@ -87,14 +87,15 @@ class ReplicationStreamer(object):
         self.streams_by_name = {stream.NAME: stream for stream in self.streams}
 
         LaterGauge(
-            "synapse_replication_tcp_resource_connections_per_stream", "", ["stream_name"],
+            "synapse_replication_tcp_resource_connections_per_stream", "",
+            ["stream_name"],
             lambda: {
                 (stream_name,): len([
                     conn for conn in self.connections
                     if stream_name in conn.replication_streams
                 ])
                 for stream_name in self.streams_by_name
-            }).register()
+            })
 
         self.federation_sender = None
         if not hs.config.send_federation:
-- 
cgit 1.4.1


From 228f1f584ef100f18440362303775f7c860c01ca Mon Sep 17 00:00:00 2001
From: Amber Brown <hawkowl@atleastfornow.net>
Date: Tue, 22 May 2018 15:02:38 -0500
Subject: fix the test failures

---
 synapse/replication/tcp/protocol.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'synapse/replication/tcp')

diff --git a/synapse/replication/tcp/protocol.py b/synapse/replication/tcp/protocol.py
index 5848f57c5e..bdb110663c 100644
--- a/synapse/replication/tcp/protocol.py
+++ b/synapse/replication/tcp/protocol.py
@@ -513,7 +513,7 @@ class ClientReplicationStreamProtocol(BaseReplicationStreamProtocol):
 
     def on_RDATA(self, cmd):
         stream_name = cmd.stream_name
-        inbound_rdata_count.inc(stream_name)
+        inbound_rdata_count.labels(stream_name).inc()
 
         try:
             row = STREAMS_MAP[stream_name].ROW_TYPE(*cmd.row)
-- 
cgit 1.4.1


From b6063631c3c5ea4c4d118cc9f9d7e2896949ee1d Mon Sep 17 00:00:00 2001
From: Amber Brown <hawkowl@atleastfornow.net>
Date: Tue, 22 May 2018 17:36:20 -0500
Subject: more cleanup

---
 synapse/metrics/__init__.py         | 4 +---
 synapse/replication/tcp/protocol.py | 9 ++++++---
 synapse/replication/tcp/resource.py | 7 ++++---
 3 files changed, 11 insertions(+), 9 deletions(-)

(limited to 'synapse/replication/tcp')

diff --git a/synapse/metrics/__init__.py b/synapse/metrics/__init__.py
index bed37b5f56..75117915ff 100644
--- a/synapse/metrics/__init__.py
+++ b/synapse/metrics/__init__.py
@@ -21,7 +21,7 @@ import platform
 import attr
 
 from prometheus_client import Gauge, Histogram, Counter
-from prometheus_client.core import GaugeMetricFamily, CounterMetricFamily, REGISTRY
+from prometheus_client.core import GaugeMetricFamily, REGISTRY
 
 from twisted.internet import reactor
 
@@ -97,8 +97,6 @@ gc_time = Histogram(
 class GCCounts(object):
 
     def collect(self):
-        gc_counts = gc.get_count()
-
         cm = GaugeMetricFamily("python_gc_counts", "GC cycle counts", labels=["gen"])
         for n, m in enumerate(gc.get_count()):
             cm.add_metric([str(n)], m)
diff --git a/synapse/replication/tcp/protocol.py b/synapse/replication/tcp/protocol.py
index bdb110663c..5995e5d553 100644
--- a/synapse/replication/tcp/protocol.py
+++ b/synapse/replication/tcp/protocol.py
@@ -197,7 +197,8 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
 
         self.last_received_command = self.clock.time_msec()
 
-        self.inbound_commands_counter[cmd_name] = self.inbound_commands_counter[cmd_name] + 1
+        self.inbound_commands_counter[cmd_name] = (
+            self.inbound_commands_counter[cmd_name] + 1)
 
         cmd_cls = COMMAND_MAP[cmd_name]
         try:
@@ -247,7 +248,8 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver):
             self._queue_command(cmd)
             return
 
-        self.outbound_commands_counter[cmd.NAME] = self.outbound_commands_counter[cmd.NAME] + 1
+        self.outbound_commands_counter[cmd.NAME] = (
+            self.outbound_commands_counter[cmd.NAME] + 1)
         string = "%s %s" % (cmd.NAME, cmd.to_line(),)
         if "\n" in string:
             raise Exception("Unexpected newline in command: %r", string)
@@ -632,4 +634,5 @@ tcp_outbound_commands = LaterGauge(
     })
 
 # number of updates received for each RDATA stream
-inbound_rdata_count = Counter("synapse_replication_tcp_inbound_rdata_count", "", ["stream_name"])
+inbound_rdata_count = Counter("synapse_replication_tcp_inbound_rdata_count", "",
+                              ["stream_name"])
diff --git a/synapse/replication/tcp/resource.py b/synapse/replication/tcp/resource.py
index 1ca30bc31a..19987c06f0 100644
--- a/synapse/replication/tcp/resource.py
+++ b/synapse/replication/tcp/resource.py
@@ -28,12 +28,13 @@ import logging
 
 from prometheus_client import Counter
 
-stream_updates_counter = Counter("synapse_replication_tcp_resource_stream_updates", "", ["stream_name"]
-)
+stream_updates_counter = Counter("synapse_replication_tcp_resource_stream_updates",
+                                 "", ["stream_name"])
 user_sync_counter = Counter("synapse_replication_tcp_resource_user_sync", "")
 federation_ack_counter = Counter("synapse_replication_tcp_resource_federation_ack", "")
 remove_pusher_counter = Counter("synapse_replication_tcp_resource_remove_pusher", "")
-invalidate_cache_counter = Counter("synapse_replication_tcp_resource_invalidate_cache", "")
+invalidate_cache_counter = Counter("synapse_replication_tcp_resource_invalidate_cache",
+                                   "")
 user_ip_cache_counter = Counter("synapse_replication_tcp_resource_user_ip_cache", "")
 
 logger = logging.getLogger(__name__)
-- 
cgit 1.4.1