diff --git a/common/http.py b/common/http.py index f708651..ca4555b 100644 --- a/common/http.py +++ b/common/http.py @@ -3,11 +3,8 @@ import httpx from .settings import settings -def downstream_client(request_id: str) -> httpx.AsyncClient: - """An httpx.AsyncClient preconfigured with the 3s downstream timeout and - the X-Request-Id header forwarded, per phase-0-base-project-spec.md ยง9 -- - this header is the only correlation mechanism that exists in Phase 0.""" - return httpx.AsyncClient( - timeout=settings.downstream_timeout_seconds, - headers={"X-Request-Id": request_id}, - ) +def downstream_client() -> httpx.AsyncClient: + """An httpx.AsyncClient preconfigured with the 3s downstream timeout. + Correlation travels as W3C `traceparent`, injected by the httpx + instrumentation (common/telemetry.py) -- it replaced Phase 0's X-Request-Id.""" + return httpx.AsyncClient(timeout=settings.downstream_timeout_seconds) diff --git a/services/fraud/app.py b/services/fraud/app.py index 5841bd3..08b8969 100644 --- a/services/fraud/app.py +++ b/services/fraud/app.py @@ -33,7 +33,6 @@ class ScoreRequest(BaseModel): holder_name: str category_id: int | None = None amount_minor: int - request_id: str = "unknown" @app.get("/") @@ -63,11 +62,11 @@ async def readyz(): @app.post("/internal/score") async def score_transaction(body: ScoreRequest): - logger.info(f"scoring account {body.account_id} request_id={body.request_id}") + logger.info(f"scoring account {body.account_id}") await redis_client.incr("fraud:calls_total") if await chaos.blocking_event_loop_enabled(): - logger.info(f"blocking_event_loop chaos engaged for request_id={body.request_id}") + logger.info(f"blocking_event_loop chaos engaged for account {body.account_id}") chaos.synchronous_cpu_work() with tracer.start_as_current_span( @@ -79,25 +78,25 @@ async def score_transaction(body: ScoreRequest): ) as span: if body.category_id is not None: span.set_attribute("ledgerline.merchant.category_id", body.category_id) - async with downstream_client(body.request_id) as client: + async with downstream_client() as client: try: resp = await client.post( f"{settings.sanctions_url}/internal/lookup", - json={"holder_name": body.holder_name, "request_id": body.request_id}, + json={"holder_name": body.holder_name}, ) except httpx.TimeoutException as exc: span.record_exception(exc) span.set_status(StatusCode.ERROR, "sanctions lookup timed out") - logger.exception(f"sanctions lookup timed out for request_id={body.request_id}") + logger.exception(f"sanctions lookup timed out for account {body.account_id}") return JSONResponse(status_code=504, content={"error": "sanctions lookup timed out"}) except httpx.RequestError as exc: span.record_exception(exc) span.set_status(StatusCode.ERROR, "sanctions lookup failed") - logger.exception(f"sanctions lookup failed for request_id={body.request_id}: {exc}") + logger.exception(f"sanctions lookup failed for account {body.account_id}: {exc}") return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"}) if resp.status_code >= 300: - logger.info(f"sanctions returned {resp.status_code} for request_id={body.request_id}") + logger.info(f"sanctions returned {resp.status_code} for account {body.account_id}") return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"}) matched = resp.json().get("matched", False) diff --git a/services/gateway/app.py b/services/gateway/app.py index 3ce85af..24603fb 100644 --- a/services/gateway/app.py +++ b/services/gateway/app.py @@ -7,6 +7,7 @@ from fastapi import FastAPI, Request from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, field_validator +from opentelemetry import trace from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.utils import suppress_instrumentation @@ -34,9 +35,7 @@ app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") # A shared, connection-pooling client for the simple read-only proxy routes # below -- opening a fresh httpx.AsyncClient (and TCP connection) per request -# is real, needless overhead under concurrent load. orchestrator.py's -# downstream_client() stays per-request since it carries per-transaction -# headers (X-Request-Id). +# is real, needless overhead under concurrent load. _proxy_client = httpx.AsyncClient( timeout=settings.downstream_timeout_seconds, limits=httpx.Limits(max_connections=100, max_keepalive_connections=50), @@ -115,14 +114,22 @@ async def account_insights(account_id: str): @app.post("/api/transactions", status_code=201) async def submit_transaction(body: TransactionRequest): - request_id = str(uuid.uuid4()) + request_id = _request_id() try: - result = await orchestrator.create_transaction(body.model_dump(), request_id) + result = await orchestrator.create_transaction(body.model_dump()) except UpstreamFailure as exc: return JSONResponse( status_code=exc.status_code, content={"error": exc.message, "request_id": request_id} ) - return JSONResponse(status_code=201, content=result) + return JSONResponse(status_code=201, content={**result, "request_id": request_id}) + + +def _request_id() -> str: + """The response's `request_id` is the trace ID: paste it into Tempo to get + the whole request. 32 hex digits, which is still a valid UUID. Falls back to + a random one if no span is recording (telemetry not set up).""" + ctx = trace.get_current_span().get_span_context() + return format(ctx.trace_id, "032x") if ctx.is_valid else uuid.uuid4().hex @app.get("/api/health/all") diff --git a/services/gateway/orchestrator.py b/services/gateway/orchestrator.py index 59f42bf..170b01a 100644 --- a/services/gateway/orchestrator.py +++ b/services/gateway/orchestrator.py @@ -74,22 +74,22 @@ async def _run_chatty_health_probe(client: httpx.AsyncClient) -> None: pass -async def create_transaction(payload: dict, request_id: str) -> dict: +async def create_transaction(payload: dict) -> dict: start = time.perf_counter() - logger.info(f"start transaction for account {payload['account_id']} request_id={request_id}") + logger.info(f"start transaction for account {payload['account_id']}") retry_storm = await chaos.retry_storm_enabled() no_batching = await chaos.no_batching_enabled() chatty_health = await chaos.chatty_health_enabled() - async with downstream_client(request_id) as client: + async with downstream_client() as client: if chatty_health: await _run_chatty_health_probe(client) create_resp = await _post( client, f"{settings.ledger_url}/internal/transactions", - {**payload, "request_id": request_id}, + payload, ) if create_resp.status_code == 404: raise UpstreamFailure(404, "account not found") @@ -105,7 +105,6 @@ async def create_transaction(payload: dict, request_id: str) -> dict: "account_id": created["account_id"], "holder_name": created["holder_name"], "category_id": created["category_id"], - "request_id": request_id, } if no_batching: @@ -125,7 +124,7 @@ async def create_transaction(payload: dict, request_id: str) -> dict: settle_resp = await _post( client, f"{settings.ledger_url}/internal/transactions/{tx_id}/{action}", - {"risk_score": risk_score, "request_id": request_id}, + {"risk_score": risk_score}, ) if settle_resp.status_code >= 300: raise UpstreamFailure(502, "ledger failed to finalise the transaction") @@ -166,10 +165,10 @@ async def create_transaction(payload: dict, request_id: str) -> dict: if summary_resp.status_code == 200: summary = summary_resp.json() except httpx.RequestError: - logger.exception(f"insights summary unavailable for request_id={request_id}") + logger.exception(f"insights summary unavailable for transaction {tx_id}") elapsed_ms = (time.perf_counter() - start) * 1000 - logger.info(f"finished transaction {tx_id} status={settled['status']} request_id={request_id}") + logger.info(f"finished transaction {tx_id} status={settled['status']}") return { "transaction_id": tx_id, @@ -177,6 +176,5 @@ async def create_transaction(payload: dict, request_id: str) -> dict: "risk_score": risk_score, "balance_minor": settled["balance_minor"], "summary": summary, - "request_id": request_id, "elapsed_ms": elapsed_ms, } diff --git a/services/ledger/app.py b/services/ledger/app.py index 187673a..99f598e 100644 --- a/services/ledger/app.py +++ b/services/ledger/app.py @@ -36,7 +36,6 @@ class CreateTransactionRequest(BaseModel): currency: str merchant_name: str category_id: int | None = None - request_id: str = "unknown" @field_validator("amount_minor") @classmethod @@ -49,7 +48,6 @@ class CreateTransactionRequest(BaseModel): class SettleRequest(BaseModel): risk_score: float - request_id: str = "unknown" @app.exception_handler(Exception) @@ -93,7 +91,7 @@ async def list_accounts(): @app.post("/internal/transactions") async def create_transaction(body: CreateTransactionRequest): - logger.info(f"processing tx for account {body.account_id}, amount {body.amount_minor} request_id={body.request_id}") + logger.info(f"processing tx for account {body.account_id}, amount {body.amount_minor}") starved = await chaos.pool_starvation_enabled() poison_enabled = await chaos.poison_data_enabled() @@ -153,7 +151,7 @@ async def settle(tx_id: str, body: SettleRequest): use_lock = row_lock and bool(account.is_hot) balance = await repository.settle_transaction(session, tx=tx, risk_score=body.risk_score, use_row_lock=use_lock) - logger.info(f"settled transaction {tx_id} request_id={body.request_id}") + logger.info(f"settled transaction {tx_id}") return {"status": "settled", "balance_minor": balance} @@ -167,5 +165,5 @@ async def reject(tx_id: str, body: SettleRequest): return JSONResponse(status_code=404, content={"error": "transaction not found"}) balance = await repository.reject_transaction(session, tx=tx, risk_score=body.risk_score) - logger.info(f"rejected transaction {tx_id} request_id={body.request_id}") + logger.info(f"rejected transaction {tx_id}") return {"status": "rejected", "balance_minor": balance} diff --git a/services/sanctions/app.py b/services/sanctions/app.py index 4d78a82..ab49ce3 100644 --- a/services/sanctions/app.py +++ b/services/sanctions/app.py @@ -30,7 +30,6 @@ SQLAlchemyInstrumentor().instrument(engines=[engine.sync_engine]) class LookupRequest(BaseModel): holder_name: str - request_id: str = "unknown" @app.get("/") @@ -61,15 +60,15 @@ async def readyz(): @app.post("/internal/lookup") async def lookup(body: LookupRequest): - logger.info(f"lookup for holder_name={body.holder_name!r} request_id={body.request_id}") + logger.info(f"lookup for holder_name={body.holder_name!r}") if await chaos.sanctions_errors_enabled() and chaos.should_error(): - logger.info(f"sanctions_errors chaos triggered for request_id={body.request_id}") + logger.info("sanctions_errors chaos triggered") return JSONResponse(status_code=503, content={"error": "sanctions vendor unavailable"}) if await chaos.slow_sanctions_enabled(): delay = chaos.sample_delay_seconds() - logger.info(f"slow_sanctions chaos sleeping {delay:.3f}s for request_id={body.request_id}") + logger.info(f"slow_sanctions chaos sleeping {delay:.3f}s") await asyncio.sleep(delay) # holder_name is PII: it stays off the span; only the outcome is recorded. diff --git a/tests/test_no_observability.py b/tests/test_no_observability.py deleted file mode 100644 index fd7c15a..0000000 --- a/tests/test_no_observability.py +++ /dev/null @@ -1,100 +0,0 @@ -"""G10 -- no observability tooling is present yet. - -This test module guards the Phase 0 "before" state. It is expected to be -DELETED at the start of Phase 2, not maintained -- once real instrumentation -is added, every assertion here becomes actively wrong on purpose. -""" -import json -import re -import subprocess -from datetime import datetime, timedelta, timezone - -import httpx -import pytest - -from harness.config import HTTP_SERVICES, REPO_ROOT, SERVICE_URLS - -pytestmark = pytest.mark.guard - -FORBIDDEN_IMPORTS = ("opentelemetry", "prometheus_client", "structlog", "sentry_sdk", "elasticapm") - - -def _python_files(): - files = [] - for name in ("services", "common"): - base = REPO_ROOT / name - if base.is_dir(): - files.extend(base.rglob("*.py")) - return files - - -def test_application_code_exists(): - """G10 (vacuous-pass guard): this module must not pass vacuously against - an empty repo -- there must be application code for the import check - below to actually inspect.""" - files = _python_files() - assert files, ( - "no .py files found under services/ or common/ -- either the app hasn't " - "been implemented yet, or the guard tests below would pass vacuously" - ) - - -def test_no_file_imports_observability_libraries(): - """G10: no file under services/ or common/ imports an observability library.""" - offenders = [] - for path in _python_files(): - text = path.read_text(errors="ignore") - for lib in FORBIDDEN_IMPORTS: - if re.search(rf"^\s*(import|from)\s+{re.escape(lib)}\b", text, re.MULTILINE): - offenders.append((str(path.relative_to(REPO_ROOT)), lib)) - assert not offenders, offenders - - -def test_dependency_files_do_not_list_observability_libraries(): - """G10: neither requirements*.txt (excluding requirements-test.txt) nor - pyproject.toml lists an observability library.""" - candidates = [p for p in REPO_ROOT.glob("requirements*.txt") if p.name != "requirements-test.txt"] - pyproject = REPO_ROOT / "pyproject.toml" - if pyproject.is_file(): - candidates.append(pyproject) - - offenders = [] - for path in candidates: - text = path.read_text(errors="ignore").lower() - for lib in FORBIDDEN_IMPORTS: - if lib in text: - offenders.append((str(path.relative_to(REPO_ROOT)), lib)) - assert not offenders, offenders - - -@pytest.mark.parametrize("service", HTTP_SERVICES) -async def test_no_service_exposes_metrics_endpoint(service): - """G10: no service exposes /metrics.""" - async with httpx.AsyncClient(timeout=5.0) as client: - resp = await client.get(f"{SERVICE_URLS[service]}/metrics") - assert resp.status_code == 404 - - -async def test_gateway_logs_are_plain_text_not_json(): - """G10: docker compose logs gateway emits plain text, not JSON: sample 20 - lines and assert none parse as a JSON object.""" - async with httpx.AsyncClient(timeout=5.0) as client: - for _ in range(3): - await client.get(f"{SERVICE_URLS['gateway']}/healthz") - - since = (datetime.now(timezone.utc) - timedelta(minutes=10)).isoformat() - result = subprocess.run( - ["docker", "compose", "logs", "gateway", "--since", since, "--no-color", "--no-log-prefix", "--tail", "20"], - cwd=REPO_ROOT, capture_output=True, text=True, timeout=15, - ) - lines = [l for l in result.stdout.splitlines() if l.strip()][-20:] - assert lines, "no recent gateway log lines found" - - json_lines = 0 - for line in lines: - try: - json.loads(line) - json_lines += 1 - except (json.JSONDecodeError, ValueError): - pass - assert json_lines == 0, f"{json_lines}/{len(lines)} gateway log lines parsed as JSON"