diff --git a/pyproject.toml b/pyproject.toml index dcbe77a..12f3cea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,9 @@ dependencies = [ ] [project.optional-dependencies] +observability = [ + "logfire>=4.32.1", +] # Full — Maximal recipe (OceanBase + all LangChain providers) + all integrations full = [ "pyobvector>=0.2.6", @@ -34,16 +37,14 @@ full = [ "langchain-community>=0.3", "langgraph>=0.2", "mcp", - "opentelemetry-api>=1.20", - "opentelemetry-sdk>=1.20", + "seekvfs[observability]", ] dev = [ "pytest>=7.4", "pytest-asyncio>=0.23", "ruff>=0.4", "mypy>=1.8", - "opentelemetry-api>=1.20", - "opentelemetry-sdk>=1.20", + "seekvfs[observability]", ] [project.urls] diff --git a/src/seekvfs/observability.py b/src/seekvfs/observability.py deleted file mode 100644 index e1deb81..0000000 --- a/src/seekvfs/observability.py +++ /dev/null @@ -1,71 +0,0 @@ -"""OpenTelemetry hooks — soft dependency. - -When ``opentelemetry-api`` is not installed, ``trace_span`` degrades to a -no-op so users never pay the instrumentation tax unless they opt in. -""" -from __future__ import annotations - -import functools -import inspect -from collections.abc import Callable -from typing import Any, TypeVar - -F = TypeVar("F", bound=Callable[..., Any]) - -_tracer: Any | None = None -_otel_available: bool | None = None - - -def _have_otel() -> bool: - global _otel_available - if _otel_available is None: - try: - import opentelemetry.trace # noqa: F401 - _otel_available = True - except ImportError: - _otel_available = False - return _otel_available - - -def get_tracer() -> Any | None: - """Lazily fetch the process-wide tracer; returns None if OT is absent.""" - global _tracer - if _tracer is not None: - return _tracer - if not _have_otel(): - return None - from opentelemetry import trace - - _tracer = trace.get_tracer("seekvfs") - return _tracer - - -def trace_span(name: str) -> Callable[[F], F]: - """Wrap a sync or async function in an OpenTelemetry span when OT is available.""" - - def decorator(fn: F) -> F: - if inspect.iscoroutinefunction(fn): - @functools.wraps(fn) - async def async_wrapper(*args: Any, **kwargs: Any) -> Any: - tracer = get_tracer() - if tracer is None: - return await fn(*args, **kwargs) - with tracer.start_as_current_span(name): - return await fn(*args, **kwargs) - - return async_wrapper # type: ignore[return-value] - else: - @functools.wraps(fn) - def sync_wrapper(*args: Any, **kwargs: Any) -> Any: - tracer = get_tracer() - if tracer is None: - return fn(*args, **kwargs) - with tracer.start_as_current_span(name): - return fn(*args, **kwargs) - - return sync_wrapper # type: ignore[return-value] - - return decorator - - -__all__ = ["trace_span", "get_tracer"] diff --git a/src/seekvfs/tools/langgraph.py b/src/seekvfs/tools/langgraph.py index 45fbe6d..823d202 100644 --- a/src/seekvfs/tools/langgraph.py +++ b/src/seekvfs/tools/langgraph.py @@ -5,7 +5,7 @@ """ from __future__ import annotations -from typing import Any, Optional +from typing import Any from pydantic import BaseModel, create_model @@ -39,7 +39,7 @@ def _build_args_schema(spec: ToolSpec) -> type[BaseModel]: fields[field_name] = (py_type, ...) else: default = field_info.get("default", None) - fields[field_name] = (Optional[py_type], default) + fields[field_name] = (py_type | None, default) return create_model(spec.name, **fields) diff --git a/src/seekvfs/vfs.py b/src/seekvfs/vfs.py index 1e4f5fb..989057a 100644 --- a/src/seekvfs/vfs.py +++ b/src/seekvfs/vfs.py @@ -13,6 +13,8 @@ """ from __future__ import annotations +from collections.abc import Iterator +from contextlib import contextmanager from typing import TYPE_CHECKING from seekvfs.exceptions import InvalidRouteConfig, VFSError @@ -23,7 +25,6 @@ RouteConfig, SearchResult, ) -from seekvfs.observability import trace_span from seekvfs.reranker import LinearReranker from seekvfs.router import Router from seekvfs.uri import SCHEME @@ -33,6 +34,18 @@ from seekvfs.tools import ToolSpecSet +@contextmanager +def _instrument_vfs(name: str) -> Iterator[None]: + try: + import logfire + except ImportError: + yield + return + + with logfire.span(name): + yield + + class VFS: """Unified front door for SeekVFS storage protocol.""" @@ -77,25 +90,24 @@ def _validate_routes(routes: dict[str, RouteConfig]) -> None: # ---------- main API ---------- - @trace_span("vfs.write") def write(self, path: str, content: bytes | str) -> None: - path = self._normalize(path) - _, route = self._router.resolve(path) - route["backend"].write(path, content) + with _instrument_vfs("vfs.write"): + path = self._normalize(path) + _, route = self._router.resolve(path) + route["backend"].write(path, content) - @trace_span("vfs.read") def read(self, path: str, hint: str | None = None) -> FileData: - path = self._normalize(path) - _, route = self._router.resolve(path) - return route["backend"].read(path, hint=hint) + with _instrument_vfs("vfs.read"): + path = self._normalize(path) + _, route = self._router.resolve(path) + return route["backend"].read(path, hint=hint) - @trace_span("vfs.read_full") def read_full(self, path: str) -> FileData: - path = self._normalize(path) - _, route = self._router.resolve(path) - return route["backend"].read_full(path) + with _instrument_vfs("vfs.read_full"): + path = self._normalize(path) + _, route = self._router.resolve(path) + return route["backend"].read_full(path) - @trace_span("vfs.search") def search( self, query: str, @@ -104,57 +116,58 @@ def search( score_threshold: float | None = None, ) -> SearchResult: """Fan out across every route sequentially, then rerank.""" - routes = self._router.all_routes() - if not routes: - return SearchResult(query=query, hits=[], searched_paths=[]) - - per_backend: list[SearchResult] = [] - for prefix, route in routes: - out = route["backend"].search( - query, - path_pattern=path_pattern, - limit=limit, - score_threshold=score_threshold, - ) - if prefix not in out.searched_paths: - out.searched_paths.append(prefix) - per_backend.append(out) + with _instrument_vfs("vfs.search"): + routes = self._router.all_routes() + if not routes: + return SearchResult(query=query, hits=[], searched_paths=[]) + + per_backend: list[SearchResult] = [] + for prefix, route in routes: + out = route["backend"].search( + query, + path_pattern=path_pattern, + limit=limit, + score_threshold=score_threshold, + ) + if prefix not in out.searched_paths: + out.searched_paths.append(prefix) + per_backend.append(out) - return self._reranker.merge(per_backend, limit=limit) + return self._reranker.merge(per_backend, limit=limit) - @trace_span("vfs.ls") def ls( self, path: str, pattern: str | None = None, recursive: bool = False, ) -> list[FileInfo]: - path = self._normalize(path) - _, route = self._router.resolve(path) - return route["backend"].ls(path, pattern=pattern, recursive=recursive) + with _instrument_vfs("vfs.ls"): + path = self._normalize(path) + _, route = self._router.resolve(path) + return route["backend"].ls(path, pattern=pattern, recursive=recursive) - @trace_span("vfs.edit") def edit(self, path: str, old: str, new: str) -> int: - path = self._normalize(path) - _, route = self._router.resolve(path) - return route["backend"].edit(path, old, new) + with _instrument_vfs("vfs.edit"): + path = self._normalize(path) + _, route = self._router.resolve(path) + return route["backend"].edit(path, old, new) - @trace_span("vfs.grep") def grep( self, pattern: str, path_pattern: str | None = None, ) -> list[GrepMatch]: - results: list[GrepMatch] = [] - for _, route in self._router.all_routes(): - results.extend(route["backend"].grep(pattern, path_pattern=path_pattern)) - return results + with _instrument_vfs("vfs.grep"): + results: list[GrepMatch] = [] + for _, route in self._router.all_routes(): + results.extend(route["backend"].grep(pattern, path_pattern=path_pattern)) + return results - @trace_span("vfs.delete") def delete(self, path: str) -> None: - path = self._normalize(path) - _, route = self._router.resolve(path) - route["backend"].delete(path) + with _instrument_vfs("vfs.delete"): + path = self._normalize(path) + _, route = self._router.resolve(path) + route["backend"].delete(path) def read_batch(self, paths: list[str]) -> dict[str, FileData]: by_backend: dict[int, tuple[object, list[str]]] = {} diff --git a/tests/unit/test_observability.py b/tests/unit/test_observability.py index d439f76..1ac0f8d 100644 --- a/tests/unit/test_observability.py +++ b/tests/unit/test_observability.py @@ -1,36 +1,50 @@ from __future__ import annotations +import builtins + import pytest from seekvfs.vfs import VFS from tests.conftest import _StubBackend -pytest.importorskip("opentelemetry") + +@pytest.fixture +def no_logfire(monkeypatch) -> None: + original_import = builtins.__import__ + + def guarded_import(name, globals=None, locals=None, fromlist=(), level=0): + if name == "logfire": + raise ModuleNotFoundError("No module named 'logfire'") + return original_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", guarded_import) @pytest.fixture -def tracer_provider(): - from opentelemetry import trace - from opentelemetry.sdk.trace import TracerProvider - from opentelemetry.sdk.trace.export import SimpleSpanProcessor - from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( - InMemorySpanExporter, +def span_exporter(): + logfire = pytest.importorskip("logfire") + testing = pytest.importorskip("logfire.testing") + + exporter = testing.TestExporter() + logfire.configure( + send_to_logfire=False, + console=False, + inspect_arguments=False, + additional_span_processors=[testing.SimpleSpanProcessor(exporter)], ) + yield exporter + logfire.shutdown() - exporter = InMemorySpanExporter() - provider = TracerProvider() - provider.add_span_processor(SimpleSpanProcessor(exporter)) - trace.set_tracer_provider(provider) - from seekvfs import observability - observability._tracer = None - observability._otel_available = None - return exporter +def test_write_emits_logfire_span(span_exporter) -> None: + vfs = VFS(routes={"seekvfs://a/": {"backend": _StubBackend()}}) + vfs.write("seekvfs://a/p", "hello") + names = [span["name"] for span in span_exporter.exported_spans_as_dict()] + assert "vfs.write" in names -def test_write_emits_span(tracer_provider) -> None: + +def test_write_read_degrade_without_logfire(no_logfire) -> None: vfs = VFS(routes={"seekvfs://a/": {"backend": _StubBackend()}}) vfs.write("seekvfs://a/p", "hello") - spans = tracer_provider.get_finished_spans() - names = {s.name for s in spans} - assert "vfs.write" in names + assert vfs.read("seekvfs://a/p").content == b"hello"