mirror of
https://git.anonymousland.org/anonymousland/synapse.git
synced 2024-09-09 13:01:53 +00:00
Make LruCache register its own metrics (#8561)
rather than have everything that instantiates an LruCache manage metrics separately, have LruCache do it itself.
This commit is contained in:
parent
da0090fdff
commit
3ee17585cd
1
changelog.d/8561.misc
Normal file
1
changelog.d/8561.misc
Normal file
@ -0,0 +1 @@
|
|||||||
|
Move metric registration code down into `LruCache`.
|
@ -34,7 +34,6 @@ from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
|
|||||||
from synapse.events import EventBase
|
from synapse.events import EventBase
|
||||||
from synapse.logging import opentracing as opentracing
|
from synapse.logging import opentracing as opentracing
|
||||||
from synapse.types import StateMap, UserID
|
from synapse.types import StateMap, UserID
|
||||||
from synapse.util.caches import register_cache
|
|
||||||
from synapse.util.caches.lrucache import LruCache
|
from synapse.util.caches.lrucache import LruCache
|
||||||
from synapse.util.metrics import Measure
|
from synapse.util.metrics import Measure
|
||||||
|
|
||||||
@ -70,8 +69,7 @@ class Auth:
|
|||||||
self.store = hs.get_datastore()
|
self.store = hs.get_datastore()
|
||||||
self.state = hs.get_state_handler()
|
self.state = hs.get_state_handler()
|
||||||
|
|
||||||
self.token_cache = LruCache(10000)
|
self.token_cache = LruCache(10000, "token_cache")
|
||||||
register_cache("cache", "token_cache", self.token_cache)
|
|
||||||
|
|
||||||
self._auth_blocking = AuthBlocking(self.hs)
|
self._auth_blocking = AuthBlocking(self.hs)
|
||||||
|
|
||||||
|
@ -20,7 +20,6 @@ from typing import Any, Dict, List, Optional, Pattern, Union
|
|||||||
|
|
||||||
from synapse.events import EventBase
|
from synapse.events import EventBase
|
||||||
from synapse.types import UserID
|
from synapse.types import UserID
|
||||||
from synapse.util.caches import register_cache
|
|
||||||
from synapse.util.caches.lrucache import LruCache
|
from synapse.util.caches.lrucache import LruCache
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@ -186,8 +185,7 @@ class PushRuleEvaluatorForEvent:
|
|||||||
|
|
||||||
|
|
||||||
# Caches (string, is_glob, word_boundary) -> regex for push. See _glob_matches
|
# Caches (string, is_glob, word_boundary) -> regex for push. See _glob_matches
|
||||||
regex_cache = LruCache(50000)
|
regex_cache = LruCache(50000, "regex_push_cache")
|
||||||
register_cache("cache", "regex_push_cache", regex_cache)
|
|
||||||
|
|
||||||
|
|
||||||
def _glob_matches(glob: str, value: str, word_boundary: bool = False) -> bool:
|
def _glob_matches(glob: str, value: str, word_boundary: bool = False) -> bool:
|
||||||
|
@ -16,7 +16,7 @@
|
|||||||
|
|
||||||
import logging
|
import logging
|
||||||
from sys import intern
|
from sys import intern
|
||||||
from typing import Callable, Dict, Optional
|
from typing import Callable, Dict, Optional, Sized
|
||||||
|
|
||||||
import attr
|
import attr
|
||||||
from prometheus_client.core import Gauge
|
from prometheus_client.core import Gauge
|
||||||
@ -92,7 +92,7 @@ class CacheMetric:
|
|||||||
def register_cache(
|
def register_cache(
|
||||||
cache_type: str,
|
cache_type: str,
|
||||||
cache_name: str,
|
cache_name: str,
|
||||||
cache,
|
cache: Sized,
|
||||||
collect_callback: Optional[Callable] = None,
|
collect_callback: Optional[Callable] = None,
|
||||||
resizable: bool = True,
|
resizable: bool = True,
|
||||||
resize_callback: Optional[Callable] = None,
|
resize_callback: Optional[Callable] = None,
|
||||||
@ -100,12 +100,15 @@ def register_cache(
|
|||||||
"""Register a cache object for metric collection and resizing.
|
"""Register a cache object for metric collection and resizing.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
cache_type
|
cache_type: a string indicating the "type" of the cache. This is used
|
||||||
|
only for deduplication so isn't too important provided it's constant.
|
||||||
cache_name: name of the cache
|
cache_name: name of the cache
|
||||||
cache: cache itself
|
cache: cache itself, which must implement __len__(), and may optionally implement
|
||||||
|
a max_size property
|
||||||
collect_callback: If given, a function which is called during metric
|
collect_callback: If given, a function which is called during metric
|
||||||
collection to update additional metrics.
|
collection to update additional metrics.
|
||||||
resizable: Whether this cache supports being resized.
|
resizable: Whether this cache supports being resized, in which case either
|
||||||
|
resize_callback must be provided, or the cache must support set_max_size().
|
||||||
resize_callback: A function which can be called to resize the cache.
|
resize_callback: A function which can be called to resize the cache.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
|
@ -24,7 +24,6 @@ from prometheus_client import Gauge
|
|||||||
from twisted.internet import defer
|
from twisted.internet import defer
|
||||||
|
|
||||||
from synapse.util.async_helpers import ObservableDeferred
|
from synapse.util.async_helpers import ObservableDeferred
|
||||||
from synapse.util.caches import register_cache
|
|
||||||
from synapse.util.caches.lrucache import LruCache
|
from synapse.util.caches.lrucache import LruCache
|
||||||
from synapse.util.caches.treecache import TreeCache, iterate_tree_cache_entry
|
from synapse.util.caches.treecache import TreeCache, iterate_tree_cache_entry
|
||||||
|
|
||||||
@ -54,10 +53,7 @@ class DeferredCache(Generic[KT, VT]):
|
|||||||
|
|
||||||
__slots__ = (
|
__slots__ = (
|
||||||
"cache",
|
"cache",
|
||||||
"name",
|
|
||||||
"keylen",
|
|
||||||
"thread",
|
"thread",
|
||||||
"metrics",
|
|
||||||
"_pending_deferred_cache",
|
"_pending_deferred_cache",
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -89,37 +85,27 @@ class DeferredCache(Generic[KT, VT]):
|
|||||||
cache_type()
|
cache_type()
|
||||||
) # type: MutableMapping[KT, CacheEntry]
|
) # type: MutableMapping[KT, CacheEntry]
|
||||||
|
|
||||||
|
def metrics_cb():
|
||||||
|
cache_pending_metric.labels(name).set(len(self._pending_deferred_cache))
|
||||||
|
|
||||||
# cache is used for completed results and maps to the result itself, rather than
|
# cache is used for completed results and maps to the result itself, rather than
|
||||||
# a Deferred.
|
# a Deferred.
|
||||||
self.cache = LruCache(
|
self.cache = LruCache(
|
||||||
max_size=max_entries,
|
max_size=max_entries,
|
||||||
keylen=keylen,
|
keylen=keylen,
|
||||||
|
cache_name=name,
|
||||||
cache_type=cache_type,
|
cache_type=cache_type,
|
||||||
size_callback=(lambda d: len(d)) if iterable else None,
|
size_callback=(lambda d: len(d)) if iterable else None,
|
||||||
evicted_callback=self._on_evicted,
|
metrics_collection_callback=metrics_cb,
|
||||||
apply_cache_factor_from_config=apply_cache_factor_from_config,
|
apply_cache_factor_from_config=apply_cache_factor_from_config,
|
||||||
)
|
)
|
||||||
|
|
||||||
self.name = name
|
|
||||||
self.keylen = keylen
|
|
||||||
self.thread = None # type: Optional[threading.Thread]
|
self.thread = None # type: Optional[threading.Thread]
|
||||||
self.metrics = register_cache(
|
|
||||||
"cache",
|
|
||||||
name,
|
|
||||||
self.cache,
|
|
||||||
collect_callback=self._metrics_collection_callback,
|
|
||||||
)
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def max_entries(self):
|
def max_entries(self):
|
||||||
return self.cache.max_size
|
return self.cache.max_size
|
||||||
|
|
||||||
def _on_evicted(self, evicted_count):
|
|
||||||
self.metrics.inc_evictions(evicted_count)
|
|
||||||
|
|
||||||
def _metrics_collection_callback(self):
|
|
||||||
cache_pending_metric.labels(self.name).set(len(self._pending_deferred_cache))
|
|
||||||
|
|
||||||
def check_thread(self):
|
def check_thread(self):
|
||||||
expected_thread = self.thread
|
expected_thread = self.thread
|
||||||
if expected_thread is None:
|
if expected_thread is None:
|
||||||
@ -154,21 +140,18 @@ class DeferredCache(Generic[KT, VT]):
|
|||||||
if val is not _Sentinel.sentinel:
|
if val is not _Sentinel.sentinel:
|
||||||
val.callbacks.update(callbacks)
|
val.callbacks.update(callbacks)
|
||||||
if update_metrics:
|
if update_metrics:
|
||||||
self.metrics.inc_hits()
|
m = self.cache.metrics
|
||||||
|
assert m # we always have a name, so should always have metrics
|
||||||
|
m.inc_hits()
|
||||||
return val.deferred
|
return val.deferred
|
||||||
|
|
||||||
val = self.cache.get(key, _Sentinel.sentinel, callbacks=callbacks)
|
val = self.cache.get(
|
||||||
if val is not _Sentinel.sentinel:
|
key, default, callbacks=callbacks, update_metrics=update_metrics
|
||||||
self.metrics.inc_hits()
|
)
|
||||||
return val
|
if val is _Sentinel.sentinel:
|
||||||
|
|
||||||
if update_metrics:
|
|
||||||
self.metrics.inc_misses()
|
|
||||||
|
|
||||||
if default is _Sentinel.sentinel:
|
|
||||||
raise KeyError()
|
raise KeyError()
|
||||||
else:
|
else:
|
||||||
return default
|
return val
|
||||||
|
|
||||||
def set(
|
def set(
|
||||||
self,
|
self,
|
||||||
|
@ -19,8 +19,6 @@ from collections import namedtuple
|
|||||||
|
|
||||||
from synapse.util.caches.lrucache import LruCache
|
from synapse.util.caches.lrucache import LruCache
|
||||||
|
|
||||||
from . import register_cache
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
@ -46,18 +44,16 @@ class DictionaryCache:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, name, max_entries=1000):
|
def __init__(self, name, max_entries=1000):
|
||||||
self.cache = LruCache(max_size=max_entries, size_callback=len)
|
self.cache = LruCache(max_size=max_entries, cache_name=name, size_callback=len)
|
||||||
|
|
||||||
self.name = name
|
self.name = name
|
||||||
self.sequence = 0
|
self.sequence = 0
|
||||||
self.thread = None
|
self.thread = None
|
||||||
# caches_by_name[name] = self.cache
|
|
||||||
|
|
||||||
class Sentinel:
|
class Sentinel:
|
||||||
__slots__ = []
|
__slots__ = []
|
||||||
|
|
||||||
self.sentinel = Sentinel()
|
self.sentinel = Sentinel()
|
||||||
self.metrics = register_cache("dictionary", name, self.cache)
|
|
||||||
|
|
||||||
def check_thread(self):
|
def check_thread(self):
|
||||||
expected_thread = self.thread
|
expected_thread = self.thread
|
||||||
@ -82,8 +78,6 @@ class DictionaryCache:
|
|||||||
"""
|
"""
|
||||||
entry = self.cache.get(key, self.sentinel)
|
entry = self.cache.get(key, self.sentinel)
|
||||||
if entry is not self.sentinel:
|
if entry is not self.sentinel:
|
||||||
self.metrics.inc_hits()
|
|
||||||
|
|
||||||
if dict_keys is None:
|
if dict_keys is None:
|
||||||
return DictionaryEntry(
|
return DictionaryEntry(
|
||||||
entry.full, entry.known_absent, dict(entry.value)
|
entry.full, entry.known_absent, dict(entry.value)
|
||||||
@ -95,7 +89,6 @@ class DictionaryCache:
|
|||||||
{k: entry.value[k] for k in dict_keys if k in entry.value},
|
{k: entry.value[k] for k in dict_keys if k in entry.value},
|
||||||
)
|
)
|
||||||
|
|
||||||
self.metrics.inc_misses()
|
|
||||||
return DictionaryEntry(False, set(), {})
|
return DictionaryEntry(False, set(), {})
|
||||||
|
|
||||||
def invalidate(self, key):
|
def invalidate(self, key):
|
||||||
|
@ -18,6 +18,7 @@ from functools import wraps
|
|||||||
from typing import Callable, Optional, Type, Union
|
from typing import Callable, Optional, Type, Union
|
||||||
|
|
||||||
from synapse.config import cache as cache_config
|
from synapse.config import cache as cache_config
|
||||||
|
from synapse.util.caches import CacheMetric, register_cache
|
||||||
from synapse.util.caches.treecache import TreeCache
|
from synapse.util.caches.treecache import TreeCache
|
||||||
|
|
||||||
|
|
||||||
@ -43,27 +44,29 @@ class _Node:
|
|||||||
|
|
||||||
class LruCache:
|
class LruCache:
|
||||||
"""
|
"""
|
||||||
Least-recently-used cache.
|
Least-recently-used cache, supporting prometheus metrics and invalidation callbacks.
|
||||||
|
|
||||||
Supports del_multi only if cache_type=TreeCache
|
Supports del_multi only if cache_type=TreeCache
|
||||||
If cache_type=TreeCache, all keys must be tuples.
|
If cache_type=TreeCache, all keys must be tuples.
|
||||||
|
|
||||||
Can also set callbacks on objects when getting/setting which are fired
|
|
||||||
when that key gets invalidated/evicted.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
max_size: int,
|
max_size: int,
|
||||||
|
cache_name: Optional[str] = None,
|
||||||
keylen: int = 1,
|
keylen: int = 1,
|
||||||
cache_type: Type[Union[dict, TreeCache]] = dict,
|
cache_type: Type[Union[dict, TreeCache]] = dict,
|
||||||
size_callback: Optional[Callable] = None,
|
size_callback: Optional[Callable] = None,
|
||||||
evicted_callback: Optional[Callable] = None,
|
metrics_collection_callback: Optional[Callable[[], None]] = None,
|
||||||
apply_cache_factor_from_config: bool = True,
|
apply_cache_factor_from_config: bool = True,
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Args:
|
Args:
|
||||||
max_size: The maximum amount of entries the cache can hold
|
max_size: The maximum amount of entries the cache can hold
|
||||||
|
|
||||||
|
cache_name: The name of this cache, for the prometheus metrics. If unset,
|
||||||
|
no metrics will be reported on this cache.
|
||||||
|
|
||||||
keylen: The length of the tuple used as the cache key. Ignored unless
|
keylen: The length of the tuple used as the cache key. Ignored unless
|
||||||
cache_type is `TreeCache`.
|
cache_type is `TreeCache`.
|
||||||
|
|
||||||
@ -73,9 +76,13 @@ class LruCache:
|
|||||||
|
|
||||||
size_callback (func(V) -> int | None):
|
size_callback (func(V) -> int | None):
|
||||||
|
|
||||||
evicted_callback (func(int)|None):
|
metrics_collection_callback:
|
||||||
if not None, called on eviction with the size of the evicted
|
metrics collection callback. This is called early in the metrics
|
||||||
entry
|
collection process, before any of the metrics registered with the
|
||||||
|
prometheus Registry are collected, so can be used to update any dynamic
|
||||||
|
metrics.
|
||||||
|
|
||||||
|
Ignored if cache_name is None.
|
||||||
|
|
||||||
apply_cache_factor_from_config (bool): If true, `max_size` will be
|
apply_cache_factor_from_config (bool): If true, `max_size` will be
|
||||||
multiplied by a cache factor derived from the homeserver config
|
multiplied by a cache factor derived from the homeserver config
|
||||||
@ -94,6 +101,19 @@ class LruCache:
|
|||||||
else:
|
else:
|
||||||
self.max_size = int(max_size)
|
self.max_size = int(max_size)
|
||||||
|
|
||||||
|
if cache_name is not None:
|
||||||
|
metrics = register_cache(
|
||||||
|
"lru_cache",
|
||||||
|
cache_name,
|
||||||
|
self,
|
||||||
|
collect_callback=metrics_collection_callback,
|
||||||
|
) # type: Optional[CacheMetric]
|
||||||
|
else:
|
||||||
|
metrics = None
|
||||||
|
|
||||||
|
# this is exposed for access from outside this class
|
||||||
|
self.metrics = metrics
|
||||||
|
|
||||||
list_root = _Node(None, None, None, None)
|
list_root = _Node(None, None, None, None)
|
||||||
list_root.next_node = list_root
|
list_root.next_node = list_root
|
||||||
list_root.prev_node = list_root
|
list_root.prev_node = list_root
|
||||||
@ -105,8 +125,8 @@ class LruCache:
|
|||||||
todelete = list_root.prev_node
|
todelete = list_root.prev_node
|
||||||
evicted_len = delete_node(todelete)
|
evicted_len = delete_node(todelete)
|
||||||
cache.pop(todelete.key, None)
|
cache.pop(todelete.key, None)
|
||||||
if evicted_callback:
|
if metrics:
|
||||||
evicted_callback(evicted_len)
|
metrics.inc_evictions(evicted_len)
|
||||||
|
|
||||||
def synchronized(f):
|
def synchronized(f):
|
||||||
@wraps(f)
|
@wraps(f)
|
||||||
@ -169,13 +189,17 @@ class LruCache:
|
|||||||
return deleted_len
|
return deleted_len
|
||||||
|
|
||||||
@synchronized
|
@synchronized
|
||||||
def cache_get(key, default=None, callbacks=[]):
|
def cache_get(key, default=None, callbacks=[], update_metrics=True):
|
||||||
node = cache.get(key, None)
|
node = cache.get(key, None)
|
||||||
if node is not None:
|
if node is not None:
|
||||||
move_node_to_front(node)
|
move_node_to_front(node)
|
||||||
node.callbacks.update(callbacks)
|
node.callbacks.update(callbacks)
|
||||||
|
if update_metrics and metrics:
|
||||||
|
metrics.inc_hits()
|
||||||
return node.value
|
return node.value
|
||||||
else:
|
else:
|
||||||
|
if update_metrics and metrics:
|
||||||
|
metrics.inc_misses()
|
||||||
return default
|
return default
|
||||||
|
|
||||||
@synchronized
|
@synchronized
|
||||||
|
@ -59,7 +59,7 @@ class LruCacheTestCase(unittest.HomeserverTestCase):
|
|||||||
self.assertEquals(cache.pop("key"), None)
|
self.assertEquals(cache.pop("key"), None)
|
||||||
|
|
||||||
def test_del_multi(self):
|
def test_del_multi(self):
|
||||||
cache = LruCache(4, 2, cache_type=TreeCache)
|
cache = LruCache(4, keylen=2, cache_type=TreeCache)
|
||||||
cache[("animal", "cat")] = "mew"
|
cache[("animal", "cat")] = "mew"
|
||||||
cache[("animal", "dog")] = "woof"
|
cache[("animal", "dog")] = "woof"
|
||||||
cache[("vehicles", "car")] = "vroom"
|
cache[("vehicles", "car")] = "vroom"
|
||||||
@ -160,7 +160,7 @@ class LruCacheCallbacksTestCase(unittest.HomeserverTestCase):
|
|||||||
m2 = Mock()
|
m2 = Mock()
|
||||||
m3 = Mock()
|
m3 = Mock()
|
||||||
m4 = Mock()
|
m4 = Mock()
|
||||||
cache = LruCache(4, 2, cache_type=TreeCache)
|
cache = LruCache(4, keylen=2, cache_type=TreeCache)
|
||||||
|
|
||||||
cache.set(("a", "1"), "value", callbacks=[m1])
|
cache.set(("a", "1"), "value", callbacks=[m1])
|
||||||
cache.set(("a", "2"), "value", callbacks=[m2])
|
cache.set(("a", "2"), "value", callbacks=[m2])
|
||||||
|
Loading…
Reference in New Issue
Block a user