OpenTelemetry telemetry as stdout JSON lines

CHANGELOG.md

@@ -7,6 +7,23 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ---
 
+## [0.5.0] — 2026-10-02
+
+### Added
+
+- OpenTelemetry telemetry using FastAPI's native support. Spans are written to stdout as one JSON object per line, so they are written to stdout for the existing log export.
+- Custom spans: `cache.historical` (HIT, MISS or STALE, memory/redis/none tier, cooldown) and `psx.fetch` (SDK function, proxied, cache-only hit).
+- Request spans carry the client address and service version. `/health` is excluded and no request headers are recorded.
+- `PSX_TELEMETRY=off` disables telemetry.
+
+### Changed
+
+- Requires `fastapi[standard]>=0.142.0,<0.143` and adds `opentelemetry-sdk>=1.45,<1.46`. Both are capped because the code uses `fastapi.telemetry` (new in 0.142) and the private `opentelemetry.sdk._logs` module.
+- Every request is traced regardless of an incoming `traceparent` sampling flag.
+- The reported `service.version` now comes from `api.__version__`, the single source of the package version.
+
+---
+
 ## [0.4.0] — 2026-09-30
 
 ### Added

api/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.5.0"

api/cache/historical.py

@@ -16,11 +16,13 @@
 from enum import Enum
 from typing import Any
 
+from opentelemetry.trace import Span
 from psxdata.exceptions import PSXRateLimitError, PSXUnavailableError
 
 from api.cache.codec import CacheEntry, decode, encode
 from api.cache.freshness import DEFAULT_MARKET_TTL, fresh_until
 from api.cache.store import TieredStore
+from api.telemetry import get_tracer
 
 LOCK_STRIPES = 256
 PSX_COOLDOWN_SECONDS = 60
@@ -51,6 +53,12 @@ def _utcnow() -> datetime:
     return datetime.now(timezone.utc)
 
 
+def _traced(span: Span, result: CacheResult, tier: str) -> CacheResult:
+    span.set_attribute("cache.status", result.status.value)
+    span.set_attribute("cache.tier", tier)
+    return result
+
+
 class HistoricalService:
     def __init__(
         self,
@@ -73,36 +81,47 @@ def __init__(
     def get(self, symbol: str, fetch: Fetcher) -> CacheResult:
         symbol = symbol.upper()
         key = cache_key(symbol)
-        entry = self._load(key)
-        if entry is not None and self._is_fresh(entry):
-            return CacheResult(entry.rows, CacheStatus.HIT, entry.fetched_at)
-
-        with self._lock_for(key):
-            entry = self._load(key)
+        with get_tracer().start_as_current_span(
+            "cache.historical", attributes={"psxdata.symbol": symbol}
+        ) as span:
+            entry, tier = self._load(key)
             if entry is not None and self._is_fresh(entry):
-                return CacheResult(entry.rows, CacheStatus.HIT, entry.fetched_at)
-            if self._in_cooldown():
-                return self._fallback(entry, PSXRateLimitError(COOLDOWN_MESSAGE))
-            try:
-                rows = fetch(symbol)
-            except PSXRateLimitError as exc:
-                self._start_cooldown()
-                return self._fallback(entry, exc)
-            except PSXUnavailableError as exc:
-                return self._fallback(entry, exc)
-
-            rows = sorted(rows, key=lambda row: row.get("date") or "", reverse=True)
-            fetched_at = self._now()
-            fresh = CacheEntry(symbol, fetched_at, fresh_until(fetched_at, self._market_ttl), rows)
-            self.store.put(key, encode(fresh))
-            return CacheResult(rows, CacheStatus.MISS, fetched_at)
+                hit = CacheResult(entry.rows, CacheStatus.HIT, entry.fetched_at)
+                return _traced(span, hit, tier)
+
+            with self._lock_for(key):
+                entry, tier = self._load(key)
+                if entry is not None and self._is_fresh(entry):
+                    hit = CacheResult(entry.rows, CacheStatus.HIT, entry.fetched_at)
+                    return _traced(span, hit, tier)
+                if self._in_cooldown():
+                    span.set_attribute("psxdata.cooldown", True)
+                    stale = self._fallback(entry, PSXRateLimitError(COOLDOWN_MESSAGE))
+                    return _traced(span, stale, tier)
+                try:
+                    rows = fetch(symbol)
+                except PSXRateLimitError as exc:
+                    self._start_cooldown()
+                    span.set_attribute("psxdata.cooldown", True)
+                    return _traced(span, self._fallback(entry, exc), tier)
+                except PSXUnavailableError as exc:
+                    return _traced(span, self._fallback(entry, exc), tier)
+
+                rows = sorted(rows, key=lambda row: row.get("date") or "", reverse=True)
+                fetched_at = self._now()
+                fresh = CacheEntry(
+                    symbol, fetched_at, fresh_until(fetched_at, self._market_ttl), rows
+                )
+                self.store.put(key, encode(fresh))
+                return _traced(span, CacheResult(rows, CacheStatus.MISS, fetched_at), "none")
 
     def close(self) -> None:
         self.store.close()
 
-    def _load(self, key: str) -> CacheEntry | None:
-        blob = self.store.get(key)
-        return decode(blob) if blob is not None else None
+    def _load(self, key: str) -> tuple[CacheEntry | None, str]:
+        blob, tier = self.store.get_with_tier(key)
+        entry = decode(blob) if blob is not None else None
+        return entry, tier if entry is not None else "none"
 
     def _is_fresh(self, entry: CacheEntry) -> bool:
         return self._now() < entry.fresh_until

api/cache/store.py

@@ -129,13 +129,18 @@ def __init__(self, l1: MemoryLRU, l2: RedisStore) -> None:
         self.l2 = l2
 
     def get(self, key: str) -> bytes | None:
+        return self.get_with_tier(key)[0]
+
+    def get_with_tier(self, key: str) -> tuple[bytes | None, str]:
+        """Like ``get``, plus which tier answered: "memory", "redis" or "none"."""
         value = self.l1.get(key)
         if value is not None:
-            return value
+            return value, "memory"
         value = self.l2.get(key)
         if value is not None:
             self.l1.set(key, value)
-        return value
+            return value, "redis"
+        return None, "none"
 
     def put(self, key: str, value: bytes) -> None:
         self.l1.set(key, value)

api/main.py

@@ -22,6 +22,7 @@
 from api.dependencies import limiter
 from api.proxy import ProxyPassthrough, ProxyUnreachableError
 from api.routers import router_registry
+from api.telemetry import TELEMETRY, ServerSpanTags
 
 logger = logging.getLogger("api")
 
@@ -35,16 +36,21 @@ async def lifespan(app: FastAPI):
     try:
         yield
     finally:
-        service.close()
+        try:
+            service.close()
+        finally:
+            TELEMETRY.flush()
 
 
 app = FastAPI(
     title="psxdata",
     lifespan=lifespan,
     servers=[{"url": "https://psxdata-api.fastapicloud.dev"}],
+    telemetry=TELEMETRY.config,
 )
 app.state.limiter = limiter
 app.add_middleware(SlowAPIMiddleware)
+app.add_middleware(ServerSpanTags)
 
 app.add_middleware(
     CORSMiddleware,

api/proxy.py

@@ -41,12 +41,15 @@
 from limits import RateLimitItemPerMinute
 from limits.storage import MemoryStorage
 from limits.strategies import MovingWindowRateLimiter
+from opentelemetry.trace import Span
 from psxdata import BaseScraper, PSXClient
 from psxdata.constants import CACHE_DIR
 from psxdata.proxy import normalize_proxy
 from psxdata.scrapers import token as token_module
 from slowapi.util import get_remote_address
 
+from api.telemetry import get_tracer
+
 PROXY_HEADER = "X-PSX-Proxy"
 ENABLE_ENV = "PSX_PROXY_PASSTHROUGH"
 ALLOWED_SCHEMES = frozenset({"http", "socks5", "socks5h"})
@@ -205,13 +208,23 @@ def proxied(self) -> bool:
         return self._proxy is not None
 
     def fetch(self, name: str, *args: Any, **kwargs: Any) -> Any:
+        with get_tracer().start_as_current_span(
+            "psx.fetch", attributes={"psxdata.function": name, "psxdata.proxied": self.proxied}
+        ) as span:
+            return self._fetch(span, name, *args, **kwargs)
+
+    def _fetch(self, span: Span, name: str, *args: Any, **kwargs: Any) -> Any:
         if self._passthrough is None or self._request is None or self._proxy is None:
             return getattr(psxdata, name)(*args, **kwargs)
         if kwargs.get("cache", True):
             try:
-                return getattr(self._passthrough.cache_only_client(), name)(*args, **kwargs)
+                result = getattr(self._passthrough.cache_only_client(), name)(*args, **kwargs)
             except CacheMiss:
                 pass
+            else:
+                span.set_attribute("psxdata.cache_only_hit", True)
+                return result
+        span.set_attribute("psxdata.cache_only_hit", False)
         with self._passthrough.acquire(self._request, self._proxy) as client:
             return getattr(client, name)(*args, **kwargs)
 

api/requirements.txt

@@ -1,6 +1,7 @@
 # API layer dependencies — installs psxdata from PyPI + FastAPI stack
 psxdata==1.1.1
-fastapi~=0.100
+fastapi[standard]>=0.142.0,<0.143
+opentelemetry-sdk>=1.45,<1.46
 uvicorn[standard]~=0.23
 slowapi~=0.1
 redis>=4.5