diff --git a/redisvl/extensions/cache/llm/_telemetry.py b/redisvl/extensions/cache/llm/_telemetry.py new file mode 100644 index 00000000..6b3e05ed --- /dev/null +++ b/redisvl/extensions/cache/llm/_telemetry.py @@ -0,0 +1,116 @@ +"""OpenTelemetry metrics for SemanticCache. Internal. + +Attributes stay low-cardinality: the cache name, the operation, the lookup +result and the error type. Prompts, filters, entry ids and distances never +become metric attributes. +""" + +import functools +import inspect +from time import perf_counter +from typing import Any, Callable, ParamSpec, TypeVar, cast + +from redisvl.query import VectorQuery +from redisvl.utils import otel +from redisvl.utils.log import get_logger + +logger = get_logger(__name__) + +P = ParamSpec("P") +R = TypeVar("R") + + +def _result(hits: list[Any]) -> str: + return "hit" if hits else "miss" + + +def _record_duration( + cache: Any, + operation: str, + start: float, + hits: list[Any] | None = None, + error: Exception | None = None, +) -> None: + # Telemetry must never change what the caller sees, so a failure here + # (an old opentelemetry-api installed by another package, say) is dropped. + try: + attributes = {otel.CACHE_NAME: cache.name, otel.CACHE_OPERATION: operation} + if error is not None: + attributes[otel.ERROR_TYPE] = otel.error_type(error) + elif hits is not None: + attributes[otel.CACHE_RESULT] = _result(hits) + otel.histogram(otel.SEMANTIC_CACHE_OPERATION_DURATION).record( + perf_counter() - start, attributes + ) + except Exception: + logger.debug("Could not record a SemanticCache metric", exc_info=True) + + +def timed( + operation: str, *, lookup: bool = False +) -> Callable[[Callable[P, R]], Callable[P, R]]: + """Record the duration of a sync or async SemanticCache method. + + A lookup also records whether it hit, judged from the hits it returns. A + failure records ``error.type`` instead and re-raises unchanged. A call that + never completes, such as a cancelled task, records nothing. + """ + + def decorate(method: Callable[P, R]) -> Callable[P, R]: + if inspect.iscoroutinefunction(method): + + @functools.wraps(method) + async def async_wrapper(cache: Any, *args: Any, **kwargs: Any) -> Any: + start = perf_counter() + try: + result = await method(cache, *args, **kwargs) + except Exception as error: + _record_duration(cache, operation, start, error=error) + raise + _record_duration( + cache, operation, start, hits=result if lookup else None + ) + return result + + return cast(Callable[P, R], async_wrapper) + + @functools.wraps(method) + def wrapper(cache: Any, *args: Any, **kwargs: Any) -> Any: + start = perf_counter() + try: + result: Any = method(cache, *args, **kwargs) + except Exception as error: + _record_duration(cache, operation, start, error=error) + raise + _record_duration(cache, operation, start, hits=result if lookup else None) + return result + + return cast(Callable[P, R], wrapper) + + return decorate + + +def record_nearest_distance( + cache_name: str, candidates: list[dict[str, Any]], hits: list[Any] +) -> None: + """Record the nearest candidate's distance, on a miss as well as a hit. + + Nothing is recorded when there was no candidate at all: an empty cache, or + a filter that excluded every entry. Candidates arrive nearest first. + """ + try: + distance = next( + ( + candidate[VectorQuery.DISTANCE_ID] + for candidate in candidates + if candidate.get(VectorQuery.DISTANCE_ID) is not None + ), + None, + ) + if distance is not None: + otel.histogram(otel.SEMANTIC_CACHE_DISTANCE).record( + float(distance), + {otel.CACHE_NAME: cache_name, otel.CACHE_RESULT: _result(hits)}, + ) + except Exception: + logger.debug("Could not record a SemanticCache metric", exc_info=True) diff --git a/redisvl/extensions/cache/llm/semantic.py b/redisvl/extensions/cache/llm/semantic.py index 781731df..fb39bd0e 100644 --- a/redisvl/extensions/cache/llm/semantic.py +++ b/redisvl/extensions/cache/llm/semantic.py @@ -3,6 +3,7 @@ from pydantic import ValidationError from redis import Redis +from redisvl.extensions.cache.llm import _telemetry from redisvl.extensions.cache.llm.base import BaseLLMCache from redisvl.extensions.cache.llm.schema import ( CacheEntry, @@ -350,6 +351,7 @@ async def aclear(self) -> None: """ await super().aclear() + @_telemetry.timed("drop") def drop(self, ids: list[str] | None = None, keys: list[str] | None = None) -> None: """Drop specific entries from the cache by ID or Redis key. @@ -374,6 +376,7 @@ def drop(self, ids: list[str] | None = None, keys: list[str] | None = None) -> N if keys is not None: self._index.drop_keys(keys) + @_telemetry.timed("drop") async def adrop( self, ids: list[str] | None = None, keys: list[str] | None = None ) -> None: @@ -428,6 +431,7 @@ def _check_vector_dims(self, vector: list[float]): ].attrs.dims # type: ignore validate_vector_dims(len(vector), schema_vector_dims) + @_telemetry.timed("check", lookup=True) def check( self, prompt: str | None = None, @@ -508,9 +512,11 @@ def check( # Refresh TTL on all found keys self._expire_many(redis_keys) + _telemetry.record_nearest_distance(self.name, candidates, cache_hits) return cache_hits + @_telemetry.timed("check", lookup=True) async def acheck( self, prompt: str | None = None, @@ -593,6 +599,7 @@ async def acheck( # Refresh TTL on all found keys await self._aexpire_many(redis_keys) + _telemetry.record_nearest_distance(self.name, candidates, cache_hits) return cache_hits @@ -696,6 +703,7 @@ def _process_cache_results( return redis_keys, cache_hits + @_telemetry.timed("store") def store( self, prompt: str, @@ -764,6 +772,7 @@ def store( # Return the key where the entry was stored return keys[0] + @_telemetry.timed("store") async def astore( self, prompt: str, @@ -834,6 +843,7 @@ async def astore( # Return the key where the entry was stored return keys[0] + @_telemetry.timed("update") def update(self, key: str, **kwargs) -> None: """Update specific fields within an existing cache entry. If no fields are passed, then only the document TTL is refreshed. @@ -875,6 +885,7 @@ def update(self, key: str, **kwargs) -> None: # Refresh TTL regardless of whether fields were updated self.expire(key) + @_telemetry.timed("update") async def aupdate(self, key: str, **kwargs) -> None: """Async update specific fields within an existing cache entry. If no fields are passed, then only the document TTL is refreshed. diff --git a/redisvl/utils/otel.py b/redisvl/utils/otel.py index 7d85f68d..1f2f443d 100644 --- a/redisvl/utils/otel.py +++ b/redisvl/utils/otel.py @@ -40,7 +40,7 @@ class SpanKind(Enum): # type: ignore[no-redef] if TYPE_CHECKING: - from opentelemetry.metrics import Meter + from opentelemetry.metrics import Histogram, Meter from opentelemetry.trace import Tracer INSTRUMENTATION_SCOPE = "redisvl" @@ -76,6 +76,27 @@ class SpanKind(Enum): # type: ignore[no-redef] GEN_AI_RETRIEVAL_TOP_K = "gen_ai.retrieval.top_k" +# Bucket advice. The GenAI duration buckets start at 10 ms, but a cache lookup +# on local Redis takes about 250 us excluding embedding, so these start at +# 100 us. Distance buckets are +# dense where cosine thresholds usually sit and coarse up to the maximum of 2. +# Applications can override both with SDK views. +_HISTOGRAMS = { + SEMANTIC_CACHE_OPERATION_DURATION: ( + "s", + "Duration of SemanticCache operations.", + (0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05) + + (0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0), + ), + SEMANTIC_CACHE_DISTANCE: ( + "1", + "Vector distance of the nearest SemanticCache candidate per lookup.", + tuple(i / 100 for i in range(1, 11)) + + (0.125, 0.15, 0.175, 0.2, 0.25, 0.3, 0.4, 0.5, 0.75, 1.0, 1.5, 2.0), + ), +} + + class _NoOpSpan(ContextDecorator): """A span that records nothing, usable as its own context manager or decorator, like ``start_as_current_span``'s result.""" @@ -140,3 +161,31 @@ def get_meter() -> "Meter": if not _enabled(): return _NoOpMeter() # type: ignore[return-value] return metrics.get_meter(INSTRUMENTATION_SCOPE, __version__, schema_url=SCHEMA_URL) + + +@lru_cache(maxsize=None) +def histogram(name: str) -> "Histogram": + """Return one of RedisVL's histograms, created once on its meter.""" + unit, description, boundaries = _HISTOGRAMS[name] + return get_meter().create_histogram( + name, + unit=unit, + description=description, + explicit_bucket_boundaries_advisory=boundaries, + ) + + +def error_type(error: BaseException) -> str: + """Name an exception's type as the SDK names it on exception events: the + qualified class name, prefixed by its module unless it is a builtin.""" + cls = type(error) + if cls.__module__ == "builtins": + return cls.__qualname__ + return f"{cls.__module__}.{cls.__qualname__}" + + +def _clear_caches() -> None: + """Forget the cached tracer, meter and instruments. For tests.""" + get_tracer.cache_clear() + get_meter.cache_clear() + histogram.cache_clear() diff --git a/tests/conftest.py b/tests/conftest.py index 37ccbc9f..2a002af1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -326,8 +326,7 @@ def reset(): metrics_internal._METER_PROVIDER_SET_ONCE = Once() metrics_internal._METER_PROVIDER = None metrics_internal._PROXY_METER_PROVIDER = metrics_internal._ProxyMeterProvider() - otel.get_tracer.cache_clear() - otel.get_meter.cache_clear() + otel._clear_caches() spans, reader = InMemorySpanExporter(), InMemoryMetricReader() tracer_provider = TracerProvider() diff --git a/tests/integration/test_semantic_cache_metrics.py b/tests/integration/test_semantic_cache_metrics.py new file mode 100644 index 00000000..ecaa604c --- /dev/null +++ b/tests/integration/test_semantic_cache_metrics.py @@ -0,0 +1,144 @@ +"""SemanticCache metrics, read back through an in-memory OpenTelemetry reader.""" + +import asyncio + +import pytest + +from redisvl.extensions.cache.llm import SemanticCache +from redisvl.query.filter import Tag +from redisvl.utils import otel +from redisvl.utils.vectorize import CustomVectorizer +from tests.conftest import skip_if_no_redis_search + +# "near paris" is about 0.02 from "paris" in cosine distance; "rome" and +# "bread" are 1.0 from "paris", and from each other. +VECTORS = { + "paris": [1.0, 0.0, 0.0, 0.0], + "near paris": [0.98, 0.2, 0.0, 0.0], + "bread": [0.0, 1.0, 0.0, 0.0], + "rome": [0.0, 0.0, 0.0, 1.0], +} + + +@pytest.fixture +def cache(client, redis_url, worker_id): + skip_if_no_redis_search(client) + cache = SemanticCache( + name=f"metrics_cache_{worker_id}", + vectorizer=CustomVectorizer( + embed=lambda text: VECTORS.get(text, [0.0, 0.0, 1.0, 0.0]) + ), + distance_threshold=0.1, + filterable_fields=[{"name": "label", "type": "tag"}], + redis_url=redis_url, + ) + yield cache + cache.delete() + + +def _points(reader, name): + """Map each point of one histogram, by its attributes, to ``(count, sum)``, + checking the unit and bucket advice every point carries.""" + unit, _, boundaries = otel._HISTOGRAMS[name] + data = reader.get_metrics_data() + points = {} + for resource_metrics in data.resource_metrics if data else []: + for scope_metrics in resource_metrics.scope_metrics: + for metric in scope_metrics.metrics: + if metric.name == name: + assert metric.unit == unit + for point in metric.data.data_points: + assert tuple(point.explicit_bounds) == boundaries + points[frozenset(point.attributes.items())] = ( + point.count, + point.sum, + ) + return points + + +async def _call(cache, method, use_async, *args, **kwargs): + if use_async: + return await getattr(cache, f"a{method}")(*args, **kwargs) + return getattr(cache, method)(*args, **kwargs) + + +@pytest.mark.parametrize("use_async", [False, True]) +async def test_operation_durations_carry_only_their_attributes( + cache, otel_sdk, use_async +): + _, reader = otel_sdk + key = await _call(cache, "store", use_async, "paris", "Paris") + await _call(cache, "check", use_async, "near paris") + await _call(cache, "check", use_async, "bread") + await _call(cache, "update", use_async, key, response="Paris, France") + with pytest.raises(TypeError, match="Return fields must be a list"): + await _call(cache, "check", use_async, "paris", return_fields="prompt") + await _call(cache, "drop", use_async, keys=[key]) + + points = _points(reader, otel.SEMANTIC_CACHE_OPERATION_DURATION) + name = (otel.CACHE_NAME, cache.name) + assert {attributes: count for attributes, (count, _) in points.items()} == { + frozenset({name, (otel.CACHE_OPERATION, "store")}): 1, + frozenset( + {name, (otel.CACHE_OPERATION, "check"), (otel.CACHE_RESULT, "hit")} + ): 1, + frozenset( + {name, (otel.CACHE_OPERATION, "check"), (otel.CACHE_RESULT, "miss")} + ): 1, + frozenset({name, (otel.CACHE_OPERATION, "update")}): 1, + frozenset( + {name, (otel.CACHE_OPERATION, "check"), (otel.ERROR_TYPE, "TypeError")} + ): 1, + frozenset({name, (otel.CACHE_OPERATION, "drop")}): 1, + } + + +@pytest.mark.parametrize("use_async", [False, True]) +async def test_nearest_distance_is_recorded_whenever_there_is_a_candidate( + cache, otel_sdk, use_async +): + _, reader = otel_sdk + await _call(cache, "check", use_async, "paris") # empty cache: no candidate + for prompt in ("paris", "rome"): + await _call( + cache, "store", use_async, prompt, prompt, filters={"label": "city"} + ) + await _call(cache, "check", use_async, "near paris", num_results=2) + await _call(cache, "check", use_async, "bread") + no_match = Tag("label") == "food" + await _call(cache, "check", use_async, "paris", filter_expression=no_match) + + points = _points(reader, otel.SEMANTIC_CACHE_DISTANCE) + name = (otel.CACHE_NAME, cache.name) + hit = points.pop(frozenset({name, (otel.CACHE_RESULT, "hit")})) + miss = points.pop(frozenset({name, (otel.CACHE_RESULT, "miss")})) + assert points == {} + assert hit == (1, pytest.approx(0.02, abs=0.005)) + assert miss == (1, pytest.approx(1.0, abs=1e-6)) + + +@pytest.mark.parametrize("use_async", [False, True]) +async def test_a_lookup_interrupted_during_the_ttl_refresh_records_nothing( + cache, otel_sdk, monkeypatch, use_async +): + _, reader = otel_sdk + await _call(cache, "store", use_async, "paris", "Paris") + + def interrupt(*args): + raise asyncio.CancelledError + + async def ainterrupt(*args): + raise asyncio.CancelledError + + if use_async: + monkeypatch.setattr(cache, "_aexpire_many", ainterrupt) + else: + monkeypatch.setattr(cache, "_expire_many", interrupt) + with pytest.raises(asyncio.CancelledError): + await _call(cache, "check", use_async, "paris") + + assert _points(reader, otel.SEMANTIC_CACHE_DISTANCE) == {} + durations = _points(reader, otel.SEMANTIC_CACHE_OPERATION_DURATION) + assert [dict(attributes)[otel.CACHE_OPERATION] for attributes in durations] == [ + "store" + ] diff --git a/tests/unit/test_cache_telemetry.py b/tests/unit/test_cache_telemetry.py new file mode 100644 index 00000000..cc64a099 --- /dev/null +++ b/tests/unit/test_cache_telemetry.py @@ -0,0 +1,36 @@ +"""SemanticCache telemetry must never change what the caller sees.""" + +import pytest + +from redisvl.extensions.cache.llm import _telemetry +from redisvl.utils import otel + + +class _Cache: + name = "cache" + + @_telemetry.timed("check", lookup=True) + def check(self, error=None): + if error: + raise error + return ["hit"] + + @_telemetry.timed("check", lookup=True) + async def acheck(self, error=None): + return self.check(error) + + +async def test_a_failing_meter_never_reaches_the_caller(monkeypatch): + # e.g. an opentelemetry-api older than 1.30 rejecting the bucket advice + def broken_histogram(name): + raise TypeError("unexpected keyword argument") + + monkeypatch.setattr(otel, "histogram", broken_histogram) + cache, error = _Cache(), KeyError("missing") + + assert cache.check() == ["hit"] + assert await cache.acheck() == ["hit"] + with pytest.raises(KeyError) as raised: + cache.check(error) + assert raised.value is error + _telemetry.record_nearest_distance("cache", [{"vector_distance": "0.1"}], []) diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index d52f6fbc..cff59134 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -50,6 +50,15 @@ def test_off_switch_silences_redisvl_despite_a_configured_sdk(otel_sdk, monkeypa assert "probe.disabled" not in _metric_scopes(reader) +def test_error_type_qualifies_all_but_builtin_exceptions(): + from redis.exceptions import ConnectionError as RedisConnectionError + + assert otel.error_type(ConnectionError()) == "ConnectionError" + assert otel.error_type(RedisConnectionError()) == ( + "redis.exceptions.ConnectionError" + ) + + def test_without_an_sdk_redisvl_configures_nothing_and_costs_microseconds(): """With the API alone, RedisVL must leave the global providers as the API's proxies, never initialise redis-py's observability, and record nothing.