diff --git a/changelog.d/20097.misc b/changelog.d/20097.misc new file mode 100644 index 0000000000..3c9cd6c86e --- /dev/null +++ b/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. diff --git a/synapse/storage/util/id_generators.py b/synapse/storage/util/id_generators.py index d294119bb2..fbf75d5a2a 100644 --- a/synapse/storage/util/id_generators.py +++ b/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 ( diff --git a/tests/storage/test_id_generators.py b/tests/storage/test_id_generators.py index a42247b498..0666eec00c 100644 --- a/tests/storage/test_id_generators.py +++ b/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. diff --git a/tests/unittest.py b/tests/unittest.py index 51d0514095..fee928bc4c 100644 --- a/tests/unittest.py +++ b/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