Fix inaccurate per-block metrics (#6491)

`Measure` incorrectly assumed that it was the only thing being done by the parent `LoggingContext`. For instance, during a "renew group attestations" operation, hundreds of `outbound_request` calls could take place in parallel, all using the same `LoggingContext`. This would mean that any resources used during *any* of those calls would be reported against *all* of them, producing wildly inaccurate results.

Instead, we now give each `Measure` block its own `LoggingContext` (using the parent `LoggingContext` mechanism to ensure that the log lines look correct and that the metrics are ultimately propogated to the top level for reporting against requests/backgrond tasks).

changelog.d/6491.bugfix

@@ -0,0 +1 @@
+Fix inaccurate per-block Prometheus metrics.

synapse/util/metrics.py

@@ -91,72 +91,48 @@ class Measure(object):
     __slots__ = [
         "clock",
         "name",
-        "start_context",
+        "_logging_context",
         "start",
-        "created_context",
-        "start_usage",
     ]
 
     def __init__(self, clock, name):
         self.clock = clock
         self.name = name
-        self.start_context = None
+        self._logging_context = None
         self.start = None
-        self.created_context = False
 
     def __enter__(self):
+        if self._logging_context:
+            raise RuntimeError("Measure() objects cannot be re-used")
+
         self.start = self.clock.time()
-        self.start_context = LoggingContext.current_context()
+        parent_context = LoggingContext.current_context()
-        if not self.start_context:
+        self._logging_context = LoggingContext(
-            self.start_context = LoggingContext("Measure")
+            "Measure[%s]" % (self.name,), parent_context
-            self.start_context.__enter__()
+        )
-            self.created_context = True
+        self._logging_context.__enter__()
-
-        self.start_usage = self.start_context.get_resource_usage()
-
         in_flight.register((self.name,), self._update_in_flight)
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        if isinstance(exc_type, Exception) or not self.start_context:
+        if not self._logging_context:
-            return
+            raise RuntimeError("Measure() block exited without being entered")
-
-        in_flight.unregister((self.name,), self._update_in_flight)
 
         duration = self.clock.time() - self.start
+        usage = self._logging_context.get_resource_usage()
 
-        block_counter.labels(self.name).inc()
+        in_flight.unregister((self.name,), self._update_in_flight)
-        block_timer.labels(self.name).inc(duration)
+        self._logging_context.__exit__(exc_type, exc_val, exc_tb)
 
-        context = LoggingContext.current_context()
-
-        if context != self.start_context:
-            logger.warning(
-                "Context has unexpectedly changed from '%s' to '%s'. (%r)",
-                self.start_context,
-                context,
-                self.name,
-            )
-            return
-
-        if not context:
-            logger.warning("Expected context. (%r)", self.name)
-            return
-
-        current = context.get_resource_usage()
-        usage = current - self.start_usage
         try:
+            block_counter.labels(self.name).inc()
+            block_timer.labels(self.name).inc(duration)
             block_ru_utime.labels(self.name).inc(usage.ru_utime)
             block_ru_stime.labels(self.name).inc(usage.ru_stime)
             block_db_txn_count.labels(self.name).inc(usage.db_txn_count)
             block_db_txn_duration.labels(self.name).inc(usage.db_txn_duration_sec)
             block_db_sched_duration.labels(self.name).inc(usage.db_sched_duration_sec)
         except ValueError:
-            logger.warning(
+            logger.warning("Failed to save metrics! Usage: %s", usage)
-                "Failed to save metrics! OLD: %r, NEW: %r", self.start_usage, current
-            )
-
-        if self.created_context:
-            self.start_context.__exit__(exc_type, exc_val, exc_tb)
 
     def _update_in_flight(self, metrics):
         """Gets called when processing in flight metrics