Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Make inflight background metrics more efficient. #7597

Merged
merged 6 commits into from
May 29, 2020
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/7597.bugfix
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix metrics failing when there is a large number of active background processes.
98 changes: 67 additions & 31 deletions synapse/metrics/background_process_metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,16 +17,18 @@
import threading
from asyncio import iscoroutine
from functools import wraps
from typing import Dict, Set
from typing import TYPE_CHECKING, Dict, Optional, Set

import six

from prometheus_client.core import REGISTRY, Counter, GaugeMetricFamily
from prometheus_client.core import REGISTRY, Counter, Gauge

from twisted.internet import defer

from synapse.logging.context import LoggingContext, PreserveLoggingContext

if TYPE_CHECKING:
import resource


logger = logging.getLogger(__name__)


Expand All @@ -36,6 +38,12 @@
["name"],
)

_background_process_in_flight_count = Gauge(
"synapse_background_process_in_flight_count",
"Number of background processes in flight",
labelnames=["name"],
)

# we set registry=None in all of these to stop them getting registered with
# the default registry. Instead we collect them all via the CustomCollector,
# which ensures that we can update them before they are collected.
Expand Down Expand Up @@ -83,11 +91,14 @@
# it's much simpler to do so than to try to combine them.)
_background_process_counts = {} # type: Dict[str, int]

# map from description to the currently running background processes.
# Set of all running background processes that have been active since the last
# time metrics were scraped.
#
# it's kept as a dict of sets rather than a big set so that we can keep track
# of process descriptions that no longer have any active processes.
_background_processes = {} # type: Dict[str, Set[_BackgroundProcess]]
# We do it like this to handle the case where we have a large number of
# background processes stacking up behind a lock or linearizer, where we then
# only need to iterate over and update metrics for the process that have
# actually been active and can ignore the idle ones.
_background_processes_active_since_last_scrape = set() # type: Set[_BackgroundProcess]

# A lock that covers the above dicts
erikjohnston marked this conversation as resolved.
Show resolved Hide resolved
_bg_metrics_lock = threading.Lock()
Expand All @@ -101,26 +112,18 @@ class _Collector(object):
"""

def collect(self):
background_process_in_flight_count = GaugeMetricFamily(
"synapse_background_process_in_flight_count",
"Number of background processes in flight",
labels=["name"],
)
global _background_processes_active_since_last_scrape

# We copy the dict so that it doesn't change from underneath us.
# We also copy the process lists as that can also change
# We swap out the _background_processes set with an empty one so that
# we can safely iterate over the set without holding the lock.
with _bg_metrics_lock:
_background_processes_copy = {
k: list(v) for k, v in six.iteritems(_background_processes)
}
_background_processes_copy = _background_processes_active_since_last_scrape
_background_processes_active_since_last_scrape = set()

for desc, processes in six.iteritems(_background_processes_copy):
background_process_in_flight_count.add_metric((desc,), len(processes))
for processes in _background_processes_copy:
for process in processes:
process.update_metrics()

yield background_process_in_flight_count

# now we need to run collect() over each of the static Counters, and
# yield each metric they return.
for m in (
Expand Down Expand Up @@ -191,13 +194,10 @@ def run():
_background_process_counts[desc] = count + 1

_background_process_start_count.labels(desc).inc()
_background_process_in_flight_count.labels(desc).inc()

with LoggingContext(desc) as context:
with BackgroundProcessLoggingContext(desc) as context:
context.request = "%s-%i" % (desc, count)
proc = _BackgroundProcess(desc, context)

with _bg_metrics_lock:
_background_processes.setdefault(desc, set()).add(proc)

try:
result = func(*args, **kwargs)
Expand All @@ -214,10 +214,7 @@ def run():
except Exception:
logger.exception("Background process '%s' threw an exception", desc)
finally:
proc.update_metrics()

with _bg_metrics_lock:
_background_processes[desc].remove(proc)
_background_process_in_flight_count.labels(desc).dec()
anoadragon453 marked this conversation as resolved.
Show resolved Hide resolved

with PreserveLoggingContext():
return run()
Expand All @@ -238,3 +235,42 @@ def wrap_as_background_process_inner_2(*args, **kwargs):
return wrap_as_background_process_inner_2

return wrap_as_background_process_inner


class BackgroundProcessLoggingContext(LoggingContext):
"""A logging context that tracks in flight metrics for background
processes.
"""

__slots__ = ["_proc"]

def __init__(self, name: str):
super().__init__(name)

self._proc = _BackgroundProcess(name, self)

def start(self, rusage: "Optional[resource._RUsage]"):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So does this run on every batch, or just when the process stops (runs out of things to process) and then starts up again?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The naming is a bit confusing here. Basically LoggingContext.{__enter__,__exit__} are called first time we enter the context and last time we leave the context, while {start,stop} get called whenever the logging context starts/stops being the active/current logging context. (The names resume and pause might have been better here)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, and "entering the context" occurs each time a background processes starts processing.

So _background_processes_active_since_last_scrape contains just those processes that have become active since the last scrape, versus those that are continuously active.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yup, exactly

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add a quick comment to where _background_process_in_flight_count is instantiated mentioning this distinction please?

"""Log context has started running (again).
"""

super().start(rusage)

# We've become active again so we make sure we're in the list of active
# procs. (Note that "start" here means we've become active, as opposed
# to starting for the first time.)
with _bg_metrics_lock:
_background_processes_active_since_last_scrape.add(self._proc)

def __exit__(self, type, value, traceback) -> None:
"""Log context has finished.
"""

super().__exit__(type, value, traceback)

# The background process has finished. We explictly remove and manually
# update the metrics here so that if nothing is scraping metrics the set
# doesn't infinitely grow.
with _bg_metrics_lock:
_background_processes_active_since_last_scrape.discard(self._proc)

self._proc.update_metrics()