Report each ID generator's current position as a metric (#20097)

When a stream advances in the database but stops being replicated to a
process, that process's view of the stream freezes. Requests that wait
for it to catch up to a token issued by another worker then time out and
return empty responses indefinitely (see #20080), and nothing exported
said so.

Report `get_current_token` from every ID generator, on every process.
The value is comparable between processes, so a stream that has stopped
reaching one of them shows up as divergence with no client traffic
needed. It is also the position that `wait_for_stream_token` waits on,
so its divergence is the failure itself rather than a proxy for it.

Being a watermark over gapless runs of persisted IDs, it also catches a
single writer of a sharded stream going quiet, which a maximum across
writers would hide behind the writers still being replicated.

changelog.d/20097.misc

@@ -0,0 +1 @@
+Add a `synapse_storage_stream_current_position` metric, reporting each stream's current position as each worker process sees it.

synapse/storage/util/id_generators.py

@@ -36,9 +36,11 @@ from typing import (
 )
 
 import attr
+from prometheus_client import Gauge
 from sortedcontainers import SortedList, SortedSet
 
 from synapse.logging import issue9533_logger
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.database import (
     DatabasePool,
@@ -55,6 +57,16 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+# We report the calculated current position as seen by this worker. Note that
+# this is always positive (even for negative streams like backfill) to make
+# monitoring/alerting easier.
+stream_current_position_gauge = Gauge(
+    "synapse_storage_stream_current_position",
+    "The stream's current position as this process sees it."
+    "Note, these are always positive, even for negative streams.",
+    labelnames=["stream_name", SERVER_NAME_LABEL],
+)
+
 
 T = TypeVar("T")
 
@@ -232,6 +244,15 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
         self._writers = writers
         self._return_factor = 1 if positive else -1
 
+        # The `instance_name` label is what separates one worker's metrics from
+        # another's when there are multiple Synapse workers for the same homeserver
+        # running in the same process. In production the scrape labels would do that
+        # too, but tests run several homeservers in one process.
+        self._current_position_gauge = stream_current_position_gauge.labels(
+            stream_name=stream_name,
+            **{SERVER_NAME_LABEL: server_name},
+        )
+
         # We lock as some functions may be called from DB threads.
         self._lock = threading.Lock()
 
@@ -340,6 +361,20 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
             # position with the current minimum.
             self._current_positions[self._instance_name] = self._persisted_upto_position
 
+        with self._lock:
+            self._report_current_position()
+
+    def _report_current_position(self) -> None:
+        """Report the current position as a metric.
+
+        Must be called with `_lock` held.
+        """
+        assert self._lock.locked()
+
+        # Note that we always report this as a positive value, so we don't
+        # multiply by the `_return_factor`.
+        self._current_position_gauge.set(self._persisted_upto_position)
+
     def _load_current_ids(
         self,
         db_conn: LoggingDatabaseConnection,
@@ -828,6 +863,8 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
                 # do.
                 break
 
+        self._report_current_position()
+
         # Hacky debug logging to attempt to trace https://github.com/element-hq/synapse/issues/19795.
         # If this is the to-device stream, and we are a writer for that stream, log some stats
         if (

tests/storage/test_id_generators.py

@@ -32,7 +32,10 @@ from synapse.storage.database import (
     LoggingTransaction,
 )
 from synapse.storage.types import Cursor
-from synapse.storage.util.id_generators import MultiWriterIdGenerator
+from synapse.storage.util.id_generators import (
+    MultiWriterIdGenerator,
+    stream_current_position_gauge,
+)
 from synapse.storage.util.sequence import (
     LocalSequenceGenerator,
     PostgresSequenceGenerator,
@@ -109,6 +112,14 @@ class MultiWriterIdGeneratorBase(HomeserverTestCase):
         )
         return self.instances[instance_name]
 
+    def _get_reported_metric_position(self) -> int:
+        """The position the metric reports for the test stream."""
+
+        return self.get_prometheus_metric_current_value(
+            stream_current_position_gauge,
+            stream_name="test_stream",
+        )
+
     def _replicate(self, instance_name: str) -> None:
         """Similate a replication event for the given instance."""
 
@@ -229,6 +240,24 @@ class MultiWriterIdGeneratorTestCase(MultiWriterIdGeneratorBase):
         self.assertEqual(id_gen.get_positions(), {"master": 8})
         self.assertEqual(id_gen.get_current_token_for_writer("master"), 8)
 
+    def test_current_position_metric(self) -> None:
+        """The reported position follows `get_current_token`."""
+
+        self._insert_rows("master", 7)
+
+        id_gen = self._create_id_generator()
+
+        self.assertEqual(self._get_reported_metric_position(), 7)
+
+        async def _get_next_async() -> None:
+            async with id_gen.get_next():
+                pass
+
+        self.get_success(_get_next_async())
+
+        self.assertEqual(id_gen.get_current_token(), 8)
+        self.assertEqual(self._get_reported_metric_position(), 8)
+
     def test_cancelled_enter_does_not_wedge_position(self) -> None:
         """Reproduces presence getting stuck.
 

tests/unittest.py

@@ -32,6 +32,7 @@ import sys
 import time
 from collections.abc import Set
 from typing import (
+    TYPE_CHECKING,
     Any,
     Awaitable,
     Callable,
@@ -76,6 +77,7 @@ from synapse.logging.context import (
     current_context,
     set_current_context,
 )
+from synapse.metrics import SERVER_NAME_LABEL
 from synapse.rest import RegisterServletsFunc
 from synapse.server import HomeServer
 from synapse.storage.keys import FetchKeyResult
@@ -95,6 +97,9 @@ from tests.test_utils import event_injection, setup_awaitable_errors
 from tests.test_utils.logging_setup import setup_logging
 from tests.utils import checked_cast, default_config, setupdb
 
+if TYPE_CHECKING:
+    from prometheus_client.registry import Collector
+
 setupdb()
 setup_logging()
 
@@ -1225,6 +1230,72 @@ class HomeserverTestCase(TestCase):
             event_injection.inject_member_event(self.hs, room, user, membership)
         )
 
+    def get_prometheus_metric_current_value(
+        self, metric: "Collector", **labels: str
+    ) -> int:
+        """Get the value of a prometheus metric with the given labels.
+
+        This function will raise an AssertionError if there is not exactly one
+        sample with the given labels.
+
+        Note that the metrics outlives each individual test, so it may hold
+        values from previous tests.
+
+        Automatically includes SERVER_NAME_LABEL.
+        """
+
+        labels = dict(labels)
+        labels[SERVER_NAME_LABEL] = self.hs.hostname
+
+        # Matching samples for the given labels.
+        found_samples = []
+
+        for collected in metric.collect():
+            for sample in collected.samples:
+                # Check that all the labels match. If any label doesn't match,
+                # we skip this sample.
+                for label, value in labels.items():
+                    if sample.labels.get(label) != value:
+                        break
+                else:
+                    # We didn't break, so all the labels matched. Return this
+                    # sample's value.
+                    found_samples.append(sample)
+
+        # The caller expects there to be exactly one sample with the given
+        # labels. If there are multiple (or zero) samples, we error.
+        if len(found_samples) == 1:
+            # We found exactly one sample with the given labels, so return its
+            # value.
+            return int(found_samples[0].value)
+        elif len(found_samples) > 1:
+            # We found multiple samples with the given labels, so we error. We
+            # helpfully include the differences in labels between the samples to
+            # help the caller figure out why they got multiple samples.
+            labels_differences_dicts = _get_dict_differences(
+                [sample.labels for sample in found_samples]
+            )
+            differences_str = "\n".join(f" {diff}" for diff in labels_differences_dicts)
+
+            raise AssertionError(
+                f"Multiple metrics found for '{metric}' with labels {labels}\n\n"
+                "`get_prometheus_metric_current_value(...)` expects you to be specific enough"
+                "with labels that only one metric matches. Either, the metrics changed and"
+                "that's wrong in and of itself or you need to update the test to be more"
+                "specific with the labels. The extra labels you can match with are:\n"
+                f"{differences_str}"
+            )
+        else:
+            all_metrics = "\n".join(
+                f" {sample.labels}"
+                for collected in metric.collect()
+                for sample in collected.samples
+            )
+            raise AssertionError(
+                f"No metric found for {metric} with labels {labels}\n"
+                f"All metrics:\n{all_metrics}"
+            )
+
 
 class FederatingHomeserverTestCase(HomeserverTestCase):
     """
@@ -1407,3 +1478,25 @@ def skip_unless(condition: bool, reason: str) -> Callable[[TV], TV]:
         return f
 
     return decorator
+
+
+def _get_dict_differences(dicts: list[dict[str, str]]) -> list[dict[str, str]]:
+    """Return a list of dicts where each dict has the common key/values removed.
+
+    Useful for printing comparisons of prometheus metrics with different labels.
+    """
+    if not dicts:
+        return []
+
+    # Find the common key/values across all dicts
+    common_items = set(dicts[0].items())
+    for d in dicts[1:]:
+        common_items.intersection_update(d.items())
+
+    # Remove the common items from each dict
+    differences = []
+    for d in dicts:
+        diff = {k: v for k, v in d.items() if (k, v) not in common_items}
+        differences.append(diff)
+
+    return differences