From c29d1474a0254b1b13aef37e4559686c6eff6cfe Mon Sep 17 00:00:00 2001 From: mtauha Date: Fri, 2 Oct 2026 13:47:16 +0500 Subject: [PATCH 1/6] feat: OpenTelemetry stdout JSON exporters and provider setup --- api/telemetry.py | 189 ++++++++++++++++++++++++++++++++++++ pyproject.toml | 3 +- tests/api/test_telemetry.py | 133 +++++++++++++++++++++++++ 3 files changed, 324 insertions(+), 1 deletion(-) create mode 100644 api/telemetry.py create mode 100644 tests/api/test_telemetry.py diff --git a/api/telemetry.py b/api/telemetry.py new file mode 100644 index 0000000..fad17b5 --- /dev/null +++ b/api/telemetry.py @@ -0,0 +1,189 @@ +"""OpenTelemetry for psxdata-api, written to stdout as one JSON object per line. + +FastAPI Cloud keeps the app's stdout as its logs, and psx-log-exporter copies those into +MotherDuck, so telemetry travels the same path as every other log line. The line format is +the contract with psx-log-exporter's views: change a key and bump LINE_VERSION. +""" +from __future__ import annotations + +import json +import os +import sys +import threading +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from importlib.metadata import PackageNotFoundError, version +from typing import Any, TextIO + +from opentelemetry import trace +from opentelemetry.sdk._logs import LoggerProvider, ReadableLogRecord +from opentelemetry.sdk._logs.export import ( + BatchLogRecordProcessor, + LogRecordExporter, + LogRecordExportResult, +) +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter, SpanExportResult + +SERVICE_NAME = "psxdata-api" +LINE_VERSION = 1 +MAX_STACKTRACE = 8192 +KILL_SWITCH_ENV = "PSX_TELEMETRY" +_TRUNCATED = "…[truncated]" + +try: + SERVICE_VERSION = version("psxdata-api") +except PackageNotFoundError: # running from a source tree without an install (Docker image) + SERVICE_VERSION = "unknown" + + +def _hex(value: int, width: int) -> str | None: + return format(value, f"0{width}x") if value else None + + +def _dumps(data: dict[str, Any]) -> str: + return json.dumps(data, separators=(",", ":"), ensure_ascii=False, default=str) + + +def span_line(span: ReadableSpan) -> str: + context = span.context + return _dumps({ + "otel": "span", + "v": LINE_VERSION, + "trace_id": _hex(context.trace_id, 32), + "span_id": _hex(context.span_id, 16), + "parent_span_id": _hex(span.parent.span_id, 16) if span.parent else None, + "name": span.name, + "kind": span.kind.name, + "start_ns": span.start_time, + "end_ns": span.end_time, + "status": span.status.status_code.name, + "attributes": dict(span.attributes or {}), + }) + + +def _log_attributes(attributes: Mapping[str, Any] | None) -> dict[str, Any]: + out = dict(attributes or {}) + stack = out.get("exception.stacktrace") + if isinstance(stack, str) and len(stack) > MAX_STACKTRACE: + out["exception.stacktrace"] = stack[:MAX_STACKTRACE] + _TRUNCATED + return out + + +def log_line(item: ReadableLogRecord) -> str: + record = item.log_record + severity = record.severity_number + return _dumps({ + "otel": "log", + "v": LINE_VERSION, + "timestamp_ns": record.timestamp or record.observed_timestamp, + "severity_text": record.severity_text, + "severity_number": severity.value if severity is not None else None, + "body": None if record.body is None else str(record.body), + "trace_id": _hex(record.trace_id or 0, 32), + "span_id": _hex(record.span_id or 0, 16), + "attributes": _log_attributes(record.attributes), + }) + + +class _LineWriter: + """Writes whole lines under a lock; ``stream=None`` means sys.stdout at write time.""" + + def __init__(self, stream: TextIO | None) -> None: + self._stream = stream + self._lock = threading.Lock() + + def write(self, lines: list[str]) -> bool: + stream = self._stream or sys.stdout + try: + with self._lock: + stream.write("".join(line + "\n" for line in lines)) + stream.flush() + except (OSError, ValueError): # closed or broken stream: drop, never break a request + return False + return True + + +class JsonLineSpanExporter(SpanExporter): + def __init__(self, stream: TextIO | None = None) -> None: + self._writer = _LineWriter(stream) + + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + ok = self._writer.write([span_line(s) for s in spans]) + return SpanExportResult.SUCCESS if ok else SpanExportResult.FAILURE + + +class JsonLineLogExporter(LogRecordExporter): + def __init__(self, stream: TextIO | None = None) -> None: + self._writer = _LineWriter(stream) + + def export(self, batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult: + ok = self._writer.write([log_line(item) for item in batch]) + return LogRecordExportResult.SUCCESS if ok else LogRecordExportResult.FAILURE + + def shutdown(self) -> None: + pass + + def force_flush(self, timeout_millis: int = 30_000) -> bool: + return True + + +@dataclass +class Telemetry: + enabled: bool + tracer_provider: TracerProvider + logger_provider: LoggerProvider + config: dict[str, Any] + tracer: trace.Tracer + + def flush(self) -> None: + self.tracer_provider.force_flush() + self.logger_provider.force_flush() + + +def _is_health(scope: Mapping[str, Any]) -> bool: + return scope.get("path") == "/health" + + +def build_telemetry( + env: Mapping[str, str], + span_exporter: SpanExporter | None = None, + log_exporter: LogRecordExporter | None = None, +) -> Telemetry: + resource = Resource.create({"service.name": SERVICE_NAME, "service.version": SERVICE_VERSION}) + tracer_provider = TracerProvider(resource=resource) + tracer_provider.add_span_processor( + BatchSpanProcessor(span_exporter or JsonLineSpanExporter()) + ) + logger_provider = LoggerProvider(resource=resource) + logger_provider.add_log_record_processor( + BatchLogRecordProcessor(log_exporter or JsonLineLogExporter()) + ) + enabled = env.get(KILL_SWITCH_ENV, "").strip().lower() != "off" + if not enabled: + return Telemetry( + False, + tracer_provider, + logger_provider, + {"tracing": False, "logs": False, "metrics": False}, + trace.NoOpTracer(), + ) + config: dict[str, Any] = { + "tracer_provider": tracer_provider, + "logger_provider": logger_provider, + "metrics": False, # request metrics are derived from spans in SQL + "operation_spans": False, # custom spans cover the parts worth timing + "auto_configure": False, # stray OTEL_* env vars must not add a second exporter + "exclude": _is_health, + } + return Telemetry( + True, tracer_provider, logger_provider, config, tracer_provider.get_tracer(SERVICE_NAME) + ) + + +TELEMETRY = build_telemetry(os.environ) + + +def get_tracer() -> trace.Tracer: + return TELEMETRY.tracer diff --git a/pyproject.toml b/pyproject.toml index a5f8f3a..951b7b2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,8 @@ requires-python = ">=3.11" license = {file = "LICENSE.md"} dependencies = [ "psxdata[socks]==1.2.0", - "fastapi[standard]>=0.100", + "fastapi[standard]>=0.142.0", + "opentelemetry-sdk>=1.45", "uvicorn[standard]>=0.23", "slowapi>=0.1", "redis>=4.5", diff --git a/tests/api/test_telemetry.py b/tests/api/test_telemetry.py new file mode 100644 index 0000000..221f0c2 --- /dev/null +++ b/tests/api/test_telemetry.py @@ -0,0 +1,133 @@ +"""Telemetry line format, stdout exporters and provider setup (api/telemetry.py).""" +from __future__ import annotations + +import io +import json + +from opentelemetry import trace +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import ( + InMemoryLogRecordExporter, + LogRecordExportResult, + SimpleLogRecordProcessor, +) +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExportResult +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry._logs import SeverityNumber + +from api import telemetry + + +def _recorded_spans(): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = provider.get_tracer("test") + with tracer.start_as_current_span("parent", kind=trace.SpanKind.SERVER) as parent: + parent.set_attribute("http.route", "/stocks/{symbol}/historical") + with tracer.start_as_current_span("child") as child: + child.set_attribute("cache.status", "HIT") + return {s.name: s for s in exporter.get_finished_spans()} + + +def _recorded_log(body="boom", attributes=None): + exporter = InMemoryLogRecordExporter() + provider = LoggerProvider() + provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) + provider.get_logger("test").emit( + body=body, + severity_text="ERROR", + severity_number=SeverityNumber.ERROR, + attributes=attributes or {}, + ) + return exporter.get_finished_logs()[0] + + +def test_span_line_shape_and_key_order(): + spans = _recorded_spans() + line = telemetry.span_line(spans["child"]) + assert "\n" not in line + data = json.loads(line) + assert list(data) == [ + "otel", "v", "trace_id", "span_id", "parent_span_id", "name", "kind", + "start_ns", "end_ns", "status", "attributes", + ] + assert line.startswith('{"otel":"span","v":1,') + assert len(data["trace_id"]) == 32 and len(data["span_id"]) == 16 + assert data["parent_span_id"] == format(spans["parent"].context.span_id, "016x") + assert data["trace_id"] == format(spans["parent"].context.trace_id, "032x") + assert data["kind"] == "INTERNAL" and data["status"] == "UNSET" + assert data["end_ns"] >= data["start_ns"] + assert data["attributes"] == {"cache.status": "HIT"} + + +def test_root_span_has_null_parent(): + data = json.loads(telemetry.span_line(_recorded_spans()["parent"])) + assert data["parent_span_id"] is None + assert data["kind"] == "SERVER" + + +def test_log_line_shape_and_key_order(): + line = telemetry.log_line(_recorded_log()) + data = json.loads(line) + assert list(data) == [ + "otel", "v", "timestamp_ns", "severity_text", "severity_number", "body", + "trace_id", "span_id", "attributes", + ] + assert line.startswith('{"otel":"log","v":1,') + assert data["severity_text"] == "ERROR" and data["severity_number"] == 17 + assert data["body"] == "boom" + assert data["trace_id"] is None and data["span_id"] is None # emitted outside a span + + +def test_log_line_truncates_long_stacktrace(): + stack = "x" * (telemetry.MAX_STACKTRACE + 500) + data = json.loads(telemetry.log_line(_recorded_log(attributes={"exception.stacktrace": stack}))) + out = data["attributes"]["exception.stacktrace"] + assert out.endswith("…[truncated]") + assert len(out) == telemetry.MAX_STACKTRACE + len("…[truncated]") + + +def test_exporters_write_one_line_per_item(): + stream = io.StringIO() + spans = list(_recorded_spans().values()) + assert telemetry.JsonLineSpanExporter(stream).export(spans) is SpanExportResult.SUCCESS + log = _recorded_log() + assert telemetry.JsonLineLogExporter(stream).export([log]) is LogRecordExportResult.SUCCESS + out = stream.getvalue().splitlines() + assert len(out) == 3 + assert [json.loads(line)["otel"] for line in out] == ["span", "span", "log"] + + +def test_exporter_survives_closed_stream(): + stream = io.StringIO() + stream.close() + spans = list(_recorded_spans().values()) + assert telemetry.JsonLineSpanExporter(stream).export(spans) is SpanExportResult.FAILURE + assert telemetry.JsonLineLogExporter(stream).export([_recorded_log()]) is ( + LogRecordExportResult.FAILURE + ) + + +def test_build_telemetry_config_when_enabled(): + t = telemetry.build_telemetry({}) + assert t.enabled is True + cfg = t.config + assert cfg["tracer_provider"] is t.tracer_provider + assert cfg["logger_provider"] is t.logger_provider + assert cfg["metrics"] is False + assert cfg["operation_spans"] is False + assert cfg["auto_configure"] is False + assert cfg["exclude"]({"path": "/health"}) is True + assert cfg["exclude"]({"path": "/stocks"}) is False + resource = t.tracer_provider.resource.attributes + assert resource["service.name"] == "psxdata-api" + assert resource["service.version"] == telemetry.SERVICE_VERSION + + +def test_build_telemetry_kill_switch(): + t = telemetry.build_telemetry({"PSX_TELEMETRY": " OFF "}) + assert t.enabled is False + assert t.config == {"tracing": False, "logs": False, "metrics": False} + assert isinstance(t.tracer, trace.NoOpTracer) From ee7e8937e241197b28f93b3077cc7374d5fdfb0b Mon Sep 17 00:00:00 2001 From: mtauha Date: Fri, 2 Oct 2026 13:52:51 +0500 Subject: [PATCH 2/6] feat: emit request spans and error logs as stdout JSON telemetry --- api/main.py | 4 +++ api/telemetry.py | 26 +++++++++++++++-- tests/api/conftest.py | 44 +++++++++++++++++++++++++++++ tests/api/test_telemetry.py | 56 ++++++++++++++++++++++++++++++++++++- 4 files changed, 127 insertions(+), 3 deletions(-) diff --git a/api/main.py b/api/main.py index e22a22d..9918c77 100644 --- a/api/main.py +++ b/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") @@ -36,15 +37,18 @@ async def lifespan(app: FastAPI): yield finally: service.close() + 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, diff --git a/api/telemetry.py b/api/telemetry.py index fad17b5..eb26db6 100644 --- a/api/telemetry.py +++ b/api/telemetry.py @@ -15,6 +15,7 @@ from importlib.metadata import PackageNotFoundError, version from typing import Any, TextIO +from fastapi.telemetry import TelemetryConfig from opentelemetry import trace from opentelemetry.sdk._logs import LoggerProvider, ReadableLogRecord from opentelemetry.sdk._logs.export import ( @@ -25,6 +26,7 @@ from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan, TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter, SpanExportResult +from starlette.types import ASGIApp, Receive, Scope, Send SERVICE_NAME = "psxdata-api" LINE_VERSION = 1 @@ -134,7 +136,7 @@ class Telemetry: enabled: bool tracer_provider: TracerProvider logger_provider: LoggerProvider - config: dict[str, Any] + config: TelemetryConfig tracer: trace.Tracer def flush(self) -> None: @@ -169,7 +171,7 @@ def build_telemetry( {"tracing": False, "logs": False, "metrics": False}, trace.NoOpTracer(), ) - config: dict[str, Any] = { + config: TelemetryConfig = { "tracer_provider": tracer_provider, "logger_provider": logger_provider, "metrics": False, # request metrics are derived from spans in SQL @@ -187,3 +189,23 @@ def build_telemetry( def get_tracer() -> trace.Tracer: return TELEMETRY.tracer + + +class ServerSpanTags: + """Adds what FastAPI's server span lacks: the client address and the service version. + + Runs inside FastAPI's telemetry layer, so the request's server span is current here. + """ + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] == "http": + span = trace.get_current_span() + if span.is_recording(): + client = scope.get("client") + if client: + span.set_attribute("client.address", client[0]) + span.set_attribute("service.version", SERVICE_VERSION) + await self.app(scope, receive, send) diff --git a/tests/api/conftest.py b/tests/api/conftest.py index c159cc0..ac81727 100644 --- a/tests/api/conftest.py +++ b/tests/api/conftest.py @@ -1,6 +1,13 @@ """Shared fixtures for API tests.""" +from dataclasses import dataclass + import pytest +from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter, SimpleLogRecordProcessor +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from api import telemetry from api.cache.historical import HistoricalService from api.cache.store import MemoryLRU, RedisStore, TieredStore from api.main import app @@ -21,3 +28,40 @@ def fresh_historical_service() -> HistoricalService: service = HistoricalService(TieredStore(MemoryLRU(), RedisStore(None))) app.state.historical_service = service return service + + +@dataclass +class OtelCapture: + spans: InMemorySpanExporter + logs: InMemoryLogRecordExporter + + def server_spans(self) -> list[ReadableSpan]: + return [s for s in self.spans.get_finished_spans() if s.kind.name == "SERVER"] + + def span(self, name: str) -> ReadableSpan: + matches = [s for s in self.spans.get_finished_spans() if s.name == name] + assert len(matches) == 1, f"expected one {name!r} span, got {len(matches)}" + return matches[0] + + def lines(self) -> list[str]: + return [telemetry.span_line(s) for s in self.spans.get_finished_spans()] + [ + telemetry.log_line(r) for r in self.logs.get_finished_logs() + ] + + +@pytest.fixture(scope="session") +def _otel_capture() -> OtelCapture: + """Attach in-memory exporters to the app's providers once (processors can't be removed).""" + capture = OtelCapture(InMemorySpanExporter(), InMemoryLogRecordExporter()) + telemetry.TELEMETRY.tracer_provider.add_span_processor(SimpleSpanProcessor(capture.spans)) + telemetry.TELEMETRY.logger_provider.add_log_record_processor( + SimpleLogRecordProcessor(capture.logs) + ) + return capture + + +@pytest.fixture +def otel(_otel_capture: OtelCapture) -> OtelCapture: + _otel_capture.spans.clear() + _otel_capture.logs.clear() + return _otel_capture diff --git a/tests/api/test_telemetry.py b/tests/api/test_telemetry.py index 221f0c2..eaa6378 100644 --- a/tests/api/test_telemetry.py +++ b/tests/api/test_telemetry.py @@ -3,8 +3,12 @@ import io import json +from unittest.mock import patch +from fastapi import FastAPI +from fastapi.testclient import TestClient from opentelemetry import trace +from opentelemetry._logs import SeverityNumber from opentelemetry.sdk._logs import LoggerProvider from opentelemetry.sdk._logs.export import ( InMemoryLogRecordExporter, @@ -14,9 +18,9 @@ from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExportResult from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from opentelemetry._logs import SeverityNumber from api import telemetry +from api.main import app def _recorded_spans(): @@ -131,3 +135,53 @@ def test_build_telemetry_kill_switch(): assert t.enabled is False assert t.config == {"tracing": False, "logs": False, "metrics": False} assert isinstance(t.tracer, trace.NoOpTracer) + + +def test_request_produces_one_tagged_server_span(otel): + with patch("psxdata.tickers", return_value=["HBL"]): + resp = TestClient(app).get("/stocks") + assert resp.status_code == 200 + (span,) = otel.server_spans() + attrs = dict(span.attributes) + assert span.name == "GET /stocks" + assert attrs["http.route"] == "/stocks" + assert attrs["http.request.method"] == "GET" + assert attrs["http.response.status_code"] == 200 + assert attrs["client.address"] == "testclient" # scope client, same source as uvicorn's log + assert attrs["service.version"] == telemetry.SERVICE_VERSION + + +def test_health_is_not_traced(otel): + assert TestClient(app).get("/health").status_code == 200 + assert otel.server_spans() == [] + + +def test_unhandled_exception_logs_linked_error(otel): + with patch("psxdata.tickers", side_effect=RuntimeError("kaboom")): + resp = TestClient(app, raise_server_exceptions=False).get("/stocks") + assert resp.status_code == 500 + (span,) = otel.server_spans() + (log,) = otel.logs.get_finished_logs() + assert log.log_record.severity_text == "ERROR" + assert log.log_record.trace_id == span.context.trace_id + assert log.log_record.attributes["exception.type"] == "RuntimeError" + + +def test_handled_4xx_produces_no_error_log(otel): + # _parse_date raises HTTPException(422): a handled error, so no telemetry log record + assert TestClient(app).get("/stocks/HBL/historical?start=nope").status_code == 422 + assert otel.logs.get_finished_logs() == () + (span,) = otel.server_spans() + assert span.attributes["http.response.status_code"] == 422 + + +def test_kill_switch_app_still_serves(): + off = telemetry.build_telemetry({"PSX_TELEMETRY": "off"}) + tiny = FastAPI(telemetry=off.config) + tiny.add_middleware(telemetry.ServerSpanTags) + + @tiny.get("/ping") + def ping() -> dict: + return {"ok": True} + + assert TestClient(tiny).get("/ping").json() == {"ok": True} From 22261cc74c54f83de3e5cf9c93128fb089cf4364 Mon Sep 17 00:00:00 2001 From: mtauha Date: Fri, 2 Oct 2026 14:01:27 +0500 Subject: [PATCH 3/6] feat: trace /historical cache lookups with status and tier --- api/cache/historical.py | 69 +++++++++++++++++++----------- api/cache/store.py | 9 +++- tests/api/test_cache_historical.py | 69 ++++++++++++++++++++++++++++++ tests/api/test_cache_store.py | 11 +++++ 4 files changed, 131 insertions(+), 27 deletions(-) diff --git a/api/cache/historical.py b/api/cache/historical.py index 055cc8e..dc1e6fc 100644 --- a/api/cache/historical.py +++ b/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 diff --git a/api/cache/store.py b/api/cache/store.py index b0c3dc4..232e97e 100644 --- a/api/cache/store.py +++ b/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) diff --git a/tests/api/test_cache_historical.py b/tests/api/test_cache_historical.py index ed09f02..2214d74 100644 --- a/tests/api/test_cache_historical.py +++ b/tests/api/test_cache_historical.py @@ -1,6 +1,7 @@ """Tests for HistoricalService: hit/miss, expiry, stale fallback, cooldown, single-flight.""" import gzip import json +import redis import threading import zlib from concurrent.futures import ThreadPoolExecutor @@ -242,3 +243,71 @@ def blocking_fetch(symbol: str): assert not blocked.done() release.set() blocked.result(5) + + +def _attrs(otel, name="cache.historical"): + return dict(otel.span(name).attributes) + + +def test_span_on_miss_then_hit(clock: Clock, otel) -> None: + service = make_service(clock) + service.get("hbl", FakeFetch()) + assert _attrs(otel) == {"psxdata.symbol": "HBL", "cache.status": "MISS", "cache.tier": "none"} + otel.spans.clear() + service.get("HBL", FakeFetch()) + assert _attrs(otel) == {"psxdata.symbol": "HBL", "cache.status": "HIT", "cache.tier": "memory"} + + +def test_span_on_stale_after_rate_limit(clock: Clock, otel) -> None: + service = make_service(clock) + service.get("HBL", FakeFetch()) + clock.advance(10 * 24 * 3600) # well past any freshness window + otel.spans.clear() + limited = FakeFetch() + limited.error = PSXRateLimitError("429") + assert service.get("HBL", limited).status is CacheStatus.STALE + assert _attrs(otel) == { + "psxdata.symbol": "HBL", "cache.status": "STALE", "cache.tier": "memory", + "psxdata.cooldown": True, + } + + +def test_span_records_error_when_nothing_cached(clock: Clock, otel) -> None: + limited = FakeFetch() + limited.error = PSXRateLimitError("429") + with pytest.raises(PSXRateLimitError): + make_service(clock).get("HBL", limited) + span = otel.span("cache.historical") + assert span.status.status_code.name == "ERROR" + assert "cache.status" not in span.attributes + + +class _DownRedis: + def get(self, key): + raise redis.ConnectionError("down") + + def set(self, key, value, ex=None): + raise redis.ConnectionError("down") + + def close(self) -> None: + pass + + +def test_cache_tier_when_redis_down(clock: Clock, otel) -> None: + service = make_service(clock, l2_client=_DownRedis()) + assert service.get("HBL", FakeFetch()).status is CacheStatus.MISS + assert _attrs(otel)["cache.tier"] == "none" + otel.spans.clear() + assert service.get("HBL", FakeFetch()).status is CacheStatus.HIT + assert _attrs(otel)["cache.tier"] == "memory" + + +def test_miss_nests_fetch_span_under_cache_span(clock: Clock, otel) -> None: + from api.telemetry import get_tracer + + def fetch(symbol: str): + with get_tracer().start_as_current_span("psx.fetch"): + return FakeFetch()(symbol) + + make_service(clock).get("HBL", fetch) + assert otel.span("psx.fetch").parent.span_id == otel.span("cache.historical").context.span_id diff --git a/tests/api/test_cache_store.py b/tests/api/test_cache_store.py index 9d65b03..fd179ef 100644 --- a/tests/api/test_cache_store.py +++ b/tests/api/test_cache_store.py @@ -142,3 +142,14 @@ def test_tiered_works_when_l2_is_down() -> None: tiered = TieredStore(MemoryLRU(), RedisStore(DownRedis())) tiered.put("k", b"v") assert tiered.get("k") == b"v" + + +def test_get_with_tier_reports_which_layer_answered(): + tiered = TieredStore(MemoryLRU(), RedisStore(fakeredis.FakeRedis())) + assert tiered.get_with_tier("k") == (None, "none") + tiered.put("k", b"v") + assert tiered.get_with_tier("k") == (b"v", "memory") + tiered.l1 = MemoryLRU() # simulate a restart: only Redis still has it + assert tiered.get_with_tier("k") == (b"v", "redis") + assert tiered.get_with_tier("k") == (b"v", "memory") # refilled L1 + assert tiered.get("k") == b"v" From 2789b6f8a894e58920bceb86013f5aa18a5295d2 Mon Sep 17 00:00:00 2001 From: mtauha Date: Fri, 2 Oct 2026 14:08:27 +0500 Subject: [PATCH 4/6] feat: trace PSX fetches, including proxied cache-only hits --- api/proxy.py | 15 +++++++++++- tests/api/test_proxy.py | 51 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 1 deletion(-) diff --git a/api/proxy.py b/api/proxy.py index 78d5315..68624df 100644 --- a/api/proxy.py +++ b/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) diff --git a/tests/api/test_proxy.py b/tests/api/test_proxy.py index 6c05262..f338cfc 100644 --- a/tests/api/test_proxy.py +++ b/tests/api/test_proxy.py @@ -374,3 +374,54 @@ def test_flag(self, value, expected): def test_default_off(self): assert ProxyPassthrough.from_env({}).enabled is False + + +# --------------------------------------------------------------------------- +# Telemetry: psx.fetch span and credential hygiene +# --------------------------------------------------------------------------- + +class TestPsxFetchSpan: + def test_direct_fetch_span(self, client, otel): + with patch("psxdata.tickers", return_value=["HBL"]): + assert client.get("/stocks").status_code == 200 + attrs = dict(otel.span("psx.fetch").attributes) + assert attrs == {"psxdata.function": "tickers", "psxdata.proxied": False} + assert otel.span("psx.fetch").parent.span_id == otel.server_spans()[0].context.span_id + + def test_proxied_miss_then_cache_only_hit(self, client, enabled, otel): + with fake_psx(ScreenerScraper, result=_screener_df()) as reached: + assert _proxied(client, "/screener").status_code == 200 + first = dict(otel.span("psx.fetch").attributes) + otel.spans.clear() + assert _proxied(client, "/screener").status_code == 200 + second = dict(otel.span("psx.fetch").attributes) + assert len(reached) == 1 # second request was served from the shared cache + assert first == { + "psxdata.function": "screener", "psxdata.proxied": True, + "psxdata.cache_only_hit": False, + } + assert second == { + "psxdata.function": "screener", "psxdata.proxied": True, + "psxdata.cache_only_hit": True, + } + + def test_fetch_error_is_recorded_and_reraised(self, client, otel): + from psxdata.exceptions import PSXUnavailableError + + with patch("psxdata.tickers", side_effect=PSXUnavailableError("down")): + assert client.get("/stocks").status_code == 503 + assert otel.span("psx.fetch").status.status_code.name == "ERROR" + + def test_proxy_credentials_never_emitted(self, make_passthrough, otel): + def refuse(host: str, port: int) -> None: + raise OSError("refused") + + make_passthrough(connector=refuse) + resp = TestClient(app, raise_server_exceptions=False).get( + "/screener", headers={PROXY_HEADER: PROXY} + ) + assert resp.status_code == 502 + assert "s3cret" not in resp.text + lines = otel.lines() + assert lines, "expected telemetry for the request" + assert not [line for line in lines if "s3cret" in line or "alice" in line] From 0005ba31738e68793a888463f6b141ffef87b206 Mon Sep 17 00:00:00 2001 From: mtauha Date: Fri, 2 Oct 2026 14:13:08 +0500 Subject: [PATCH 5/6] chore: release 0.5.0 with OpenTelemetry stdout telemetry --- CHANGELOG.md | 15 +++++++++++++++ api/requirements.txt | 3 ++- pyproject.toml | 2 +- 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fd05321..f7eda1b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ 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 reach 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` and adds `opentelemetry-sdk>=1.45`. + +--- + ## [0.4.0] — 2026-09-30 ### Added diff --git a/api/requirements.txt b/api/requirements.txt index 1586c4c..88021f1 100644 --- a/api/requirements.txt +++ b/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 +opentelemetry-sdk>=1.45 uvicorn[standard]~=0.23 slowapi~=0.1 redis>=4.5 diff --git a/pyproject.toml b/pyproject.toml index 951b7b2..3b79692 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "psxdata-api" -version = "0.4.0" +version = "0.5.0" description = "FastAPI REST service for Pakistan Stock Exchange (PSX) data" readme = "README.md" requires-python = ">=3.11" From ebcba453fa7ab9b120fdad80b72d5a51b06b43a6 Mon Sep 17 00:00:00 2001 From: mtauha Date: Fri, 2 Oct 2026 18:33:29 +0500 Subject: [PATCH 6/6] fix: always sample, version from code, bounded otel deps Drop tests/api/__init__.py: it made pytest import the tests directory as the api package, which hid api.__version__. --- CHANGELOG.md | 6 ++++-- api/__init__.py | 1 + api/main.py | 6 ++++-- api/requirements.txt | 4 ++-- api/telemetry.py | 15 +++++++++------ pyproject.toml | 9 ++++++--- tests/api/__init__.py | 0 tests/api/test_telemetry.py | 23 +++++++++++++++++++++++ 8 files changed, 49 insertions(+), 15 deletions(-) delete mode 100644 tests/api/__init__.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f7eda1b..9f162cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,14 +11,16 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Added -- OpenTelemetry telemetry using FastAPI's native support. Spans are written to stdout as one JSON object per line, so they reach the existing log export. +- 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` and adds `opentelemetry-sdk>=1.45`. +- 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. --- diff --git a/api/__init__.py b/api/__init__.py index e69de29..3d18726 100644 --- a/api/__init__.py +++ b/api/__init__.py @@ -0,0 +1 @@ +__version__ = "0.5.0" diff --git a/api/main.py b/api/main.py index 9918c77..9662f25 100644 --- a/api/main.py +++ b/api/main.py @@ -36,8 +36,10 @@ async def lifespan(app: FastAPI): try: yield finally: - service.close() - TELEMETRY.flush() + try: + service.close() + finally: + TELEMETRY.flush() app = FastAPI( diff --git a/api/requirements.txt b/api/requirements.txt index 88021f1..9c72ece 100644 --- a/api/requirements.txt +++ b/api/requirements.txt @@ -1,7 +1,7 @@ # API layer dependencies — installs psxdata from PyPI + FastAPI stack psxdata==1.1.1 -fastapi[standard]>=0.142.0 -opentelemetry-sdk>=1.45 +fastapi[standard]>=0.142.0,<0.143 +opentelemetry-sdk>=1.45,<1.46 uvicorn[standard]~=0.23 slowapi~=0.1 redis>=4.5 diff --git a/api/telemetry.py b/api/telemetry.py index eb26db6..e772021 100644 --- a/api/telemetry.py +++ b/api/telemetry.py @@ -12,7 +12,6 @@ import threading from collections.abc import Mapping, Sequence from dataclasses import dataclass -from importlib.metadata import PackageNotFoundError, version from typing import Any, TextIO from fastapi.telemetry import TelemetryConfig @@ -26,18 +25,18 @@ from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan, TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter, SpanExportResult +from opentelemetry.sdk.trace.sampling import ALWAYS_ON from starlette.types import ASGIApp, Receive, Scope, Send +from api import __version__ + SERVICE_NAME = "psxdata-api" LINE_VERSION = 1 MAX_STACKTRACE = 8192 KILL_SWITCH_ENV = "PSX_TELEMETRY" _TRUNCATED = "…[truncated]" -try: - SERVICE_VERSION = version("psxdata-api") -except PackageNotFoundError: # running from a source tree without an install (Docker image) - SERVICE_VERSION = "unknown" +SERVICE_VERSION = __version__ def _hex(value: int, width: int) -> str | None: @@ -49,6 +48,7 @@ def _dumps(data: dict[str, Any]) -> str: def span_line(span: ReadableSpan) -> str: + """Span events and links are intentionally not serialized: exception events carry text.""" context = span.context return _dumps({ "otel": "span", @@ -70,6 +70,9 @@ def _log_attributes(attributes: Mapping[str, Any] | None) -> dict[str, Any]: stack = out.get("exception.stacktrace") if isinstance(stack, str) and len(stack) > MAX_STACKTRACE: out["exception.stacktrace"] = stack[:MAX_STACKTRACE] + _TRUNCATED + message = out.get("exception.message") + if isinstance(message, str) and len(message) > MAX_STACKTRACE: + out["exception.message"] = message[:MAX_STACKTRACE] + _TRUNCATED return out @@ -154,7 +157,7 @@ def build_telemetry( log_exporter: LogRecordExporter | None = None, ) -> Telemetry: resource = Resource.create({"service.name": SERVICE_NAME, "service.version": SERVICE_VERSION}) - tracer_provider = TracerProvider(resource=resource) + tracer_provider = TracerProvider(resource=resource, sampler=ALWAYS_ON) tracer_provider.add_span_processor( BatchSpanProcessor(span_exporter or JsonLineSpanExporter()) ) diff --git a/pyproject.toml b/pyproject.toml index 3b79692..0ef307f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,15 +4,15 @@ build-backend = "setuptools.build_meta" [project] name = "psxdata-api" -version = "0.5.0" +dynamic = ["version"] description = "FastAPI REST service for Pakistan Stock Exchange (PSX) data" readme = "README.md" requires-python = ">=3.11" license = {file = "LICENSE.md"} dependencies = [ "psxdata[socks]==1.2.0", - "fastapi[standard]>=0.142.0", - "opentelemetry-sdk>=1.45", + "fastapi[standard]>=0.142.0,<0.143", + "opentelemetry-sdk>=1.45,<1.46", "uvicorn[standard]>=0.23", "slowapi>=0.1", "redis>=4.5", @@ -39,6 +39,9 @@ Homepage = "https://github.com/psxdata/psxdata-api" Source = "https://github.com/psxdata/psxdata-api" "Bug Tracker" = "https://github.com/psxdata/psxdata-api/issues" +[tool.setuptools.dynamic] +version = {attr = "api.__version__"} + [tool.setuptools.packages.find] where = ["."] include = ["api*"] diff --git a/tests/api/__init__.py b/tests/api/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/api/test_telemetry.py b/tests/api/test_telemetry.py index eaa6378..b6e14da 100644 --- a/tests/api/test_telemetry.py +++ b/tests/api/test_telemetry.py @@ -185,3 +185,26 @@ def ping() -> dict: return {"ok": True} assert TestClient(tiny).get("/ping").json() == {"ok": True} + + +def test_unsampled_client_traceparent_is_still_recorded(otel): + headers = {"traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-00"} + with patch("psxdata.tickers", return_value=["HBL"]): + resp = TestClient(app).get("/stocks", headers=headers) + assert resp.status_code == 200 + (span,) = otel.server_spans() + assert span.context.trace_flags.sampled + + +def test_service_version_comes_from_the_package(): + import api + + assert telemetry.SERVICE_VERSION == api.__version__ + + +def test_log_line_truncates_long_exception_message(): + message = "m" * (telemetry.MAX_STACKTRACE + 500) + data = json.loads(telemetry.log_line(_recorded_log(attributes={"exception.message": message}))) + out = data["attributes"]["exception.message"] + assert out.endswith("…[truncated]") + assert len(out) == telemetry.MAX_STACKTRACE + len("…[truncated]")