mohsen_hassani/LedgerLine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace X-Request-Id with the trace ID and retire the guard suite

Browse Source 
Step 6: downstream_client() no longer sends X-Request-Id, and request_id
is gone from the internal request bodies, models, function signatures
and log lines; correlation now travels as traceparent. The gateway's
public request_id stays (the contract tests require it) but is now the
trace ID, 32 hex digits and still a valid UUID, so the value the
frontend shows can be pasted into Tempo.

Step 7: delete tests/test_no_observability.py. It guarded the Phase 0
"no observability" state, which Phase 3 ends by design.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Movses Aslanyan
2026-10-13 19:48:27 +03:30
co-authored by Claude Opus 5.5
parent 1ab425dc44
commit 7ddb614ded
7 changed files with 38 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/fraud/app.py
+7 -8
View File
@@ -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)
services/gateway/app.py
+13 -6
View File
@@ -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")
services/gateway/orchestrator.py
+7 -9
View File
@@ -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,
}
services/ledger/app.py
+3 -5
View File
@@ -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}
services/sanctions/app.py
+3 -4
View File
@@ -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.