infra/plba
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

html рендер логов

Browse Source
This commit is contained in:
zosimovaa
2026-04-30 09:56:42 +03:00
parent a144fd2912
commit fa314bc1e5
4 changed files with 369 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app_runtime/control/base.py
+1 -1
View File
@@ -19,7 +19,7 @@ class ControlActionRequest:
ActionResult = str | dict[str, object]
ActionHandler = Callable[[ControlActionRequest], Awaitable[ActionResult]]
HealthHandler = Callable[[], Awaitable[HealthPayload]]
TraceResponseFormat = Literal["json", "text"]
TraceResponseFormat = Literal["json", "text", "html"]
@dataclass(slots=True)
src/app_runtime/control/http_app.py
+14 -68
View File
@@ -1,22 +1,30 @@
from __future__ import annotations
import json
import logging
import time
from collections.abc import Awaitable, Callable
from typing import cast
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, PlainTextResponse
from fastapi.responses import JSONResponse
from app_runtime.control.base import ControlActionRequest, TraceQueryRequest
from app_runtime.contracts.trace import TraceLevel, TraceLogView
from app_runtime.control.trace_presenter import TraceRequestParser, TraceResponseRenderer
from app_runtime.contracts.trace import TraceLogView
from app_runtime.core.types import HealthPayload
LOGGER = logging.getLogger(__name__)
class HttpControlAppFactory:
def __init__(
self,
*,
trace_request_parser: TraceRequestParser | None = None,
trace_response_renderer: TraceResponseRenderer | None = None,
) -> None:
self._trace_request_parser = trace_request_parser or TraceRequestParser()
self._trace_response_renderer = trace_response_renderer or TraceResponseRenderer()
def create(
self,
health_provider: Callable[[], Awaitable[HealthPayload]],
@@ -55,7 +63,7 @@ class HttpControlAppFactory:
if trace_provider is None:
return JSONResponse(content={"status": "error", "detail": "trace lookup is not configured"}, status_code=503)
try:
trace_request = self._trace_request(request)
trace_request = self._trace_request_parser.parse(request)
except ValueError as exc:
return JSONResponse(content={"status": "error", "detail": str(exc)}, status_code=400)
try:
@@ -64,7 +72,7 @@ class HttpControlAppFactory:
return JSONResponse(content={"status": "error", "detail": f"trace not found: {traceid}"}, status_code=404)
except RuntimeError as exc:
return JSONResponse(content={"status": "error", "detail": str(exc)}, status_code=503)
return self._trace_response(payload, trace_request)
return self._trace_response_renderer.render(payload, trace_request)
return app
@@ -106,65 +114,3 @@ class HttpControlAppFactory:
if value < 0:
raise ValueError(f"query parameter must be >= 0: {name}={raw_value}")
return value
def _trace_request(self, request: Request) -> TraceQueryRequest:
raw_levels = request.query_params.get("levels")
raw_format = request.query_params.get("format", "text")
response_format = raw_format.strip().lower()
if response_format not in {"json", "text"}:
raise ValueError(f"unsupported trace format: {raw_format}")
return TraceQueryRequest(
levels=self._trace_levels(raw_levels),
include_attrs_json=self._bool_param(request, "attrs_json") or False,
response_format=response_format,
)
def _trace_levels(self, raw_levels: str | None) -> tuple[TraceLevel, ...]:
if raw_levels is None:
return ("ERROR", "WARNING", "INFO")
parts = [item.strip().upper() for item in raw_levels.split(",")]
levels = tuple(item for item in parts if item)
if not levels:
raise ValueError("trace levels must not be empty")
unsupported = [level for level in levels if level not in {"DEBUG", "INFO", "WARNING", "ERROR"}]
if unsupported:
raise ValueError(f"unsupported trace levels: {', '.join(unsupported)}")
return cast(tuple[TraceLevel, ...], levels)
def _trace_response(self, trace_view: TraceLogView, request: TraceQueryRequest) -> JSONResponse | PlainTextResponse:
if request.response_format == "json":
return JSONResponse(
content={
"trace_id": trace_view.trace_id,
"parent_id": trace_view.parent_id or "",
"child_ids": list(trace_view.child_ids),
"messages": [record.as_dict(include_attrs_json=request.include_attrs_json) for record in trace_view.records],
}
)
lines = [
f"trace_id: {trace_view.trace_id}",
f"parent_id: {trace_view.parent_id or ''}",
]
lines.extend(self._child_id_lines(trace_view.child_ids))
lines.append("--------------------------------------------------")
previous_step: str | None = None
for record in trace_view.records:
current_step = str(record.step or "")
if previous_step is None:
lines.append(f"step: {current_step}")
elif current_step != previous_step:
lines.append("--------------------------------------------------")
lines.append(f"step: {current_step}")
previous_step = current_step
line = record.message
if request.include_attrs_json:
line = f"{line}, {json.dumps(record.attrs_json, ensure_ascii=False, separators=(',', ':'))}"
lines.append(line)
return PlainTextResponse(content="\n".join(lines))
def _child_id_lines(self, child_ids: tuple[str, ...]) -> list[str]:
lines = ["child_ids:"]
if not child_ids:
return lines
lines.extend(f" - {child_id}" for child_id in child_ids)
return lines
src/app_runtime/control/trace_presenter.py
+323
View File
@@ -0,0 +1,323 @@
from __future__ import annotations
import json
from dataclasses import dataclass
from html import escape
from urllib.parse import urlencode
from fastapi import Request
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response
from app_runtime.control.base import TraceQueryRequest
from app_runtime.contracts.trace import TraceLevel, TraceLogRecord, TraceLogView
@dataclass(frozen=True)
class TraceStepGroup:
step: str
records: tuple[TraceLogRecord, ...]
class TraceRequestParser:
def parse(self, request: Request) -> TraceQueryRequest:
raw_levels = request.query_params.get("levels")
raw_format = request.query_params.get("format", "text")
response_format = raw_format.strip().lower()
if response_format not in {"json", "text", "html"}:
raise ValueError(f"unsupported trace format: {raw_format}")
return TraceQueryRequest(
levels=self._trace_levels(raw_levels),
include_attrs_json=self._bool_param(request, "attrs_json") or False,
response_format=response_format,
)
def _trace_levels(self, raw_levels: str | None) -> tuple[TraceLevel, ...]:
if raw_levels is None:
return ("ERROR", "WARNING", "INFO")
parts = [item.strip().upper() for item in raw_levels.split(",")]
levels = tuple(item for item in parts if item)
if not levels:
raise ValueError("trace levels must not be empty")
unsupported = [level for level in levels if level not in {"DEBUG", "INFO", "WARNING", "ERROR"}]
if unsupported:
raise ValueError(f"unsupported trace levels: {', '.join(unsupported)}")
return levels
def _bool_param(self, request: Request, name: str) -> bool | None:
raw_value = request.query_params.get(name)
if raw_value is None:
return None
normalized = raw_value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
raise ValueError(f"invalid boolean query parameter: {name}={raw_value}")
class TraceResponseRenderer:
def render(self, trace_view: TraceLogView, request: TraceQueryRequest) -> Response:
if request.response_format == "json":
return self._render_json(trace_view, request)
if request.response_format == "html":
return self._render_html(trace_view, request)
return self._render_text(trace_view, request)
def _render_json(self, trace_view: TraceLogView, request: TraceQueryRequest) -> JSONResponse:
return JSONResponse(
content={
"trace_id": trace_view.trace_id,
"parent_id": trace_view.parent_id or "",
"child_ids": list(trace_view.child_ids),
"messages": [record.as_dict(include_attrs_json=request.include_attrs_json) for record in trace_view.records],
}
)
def _render_text(self, trace_view: TraceLogView, request: TraceQueryRequest) -> PlainTextResponse:
lines = [
f"trace_id: {trace_view.trace_id}",
f"parent_id: {trace_view.parent_id or ''}",
*self._child_id_lines(trace_view.child_ids),
"--------------------------------------------------",
]
previous_step: str | None = None
for record in trace_view.records:
current_step = str(record.step or "")
if previous_step is None:
lines.append(f"step: {current_step}")
elif current_step != previous_step:
lines.append("--------------------------------------------------")
lines.append(f"step: {current_step}")
previous_step = current_step
lines.append(self._text_message(record, request.include_attrs_json))
return PlainTextResponse(content="\n".join(lines))
def _render_html(self, trace_view: TraceLogView, request: TraceQueryRequest) -> HTMLResponse:
title = escape(f"Trace {trace_view.trace_id}")
header_links = self._related_trace_links(trace_view, request)
rows = self._step_rows(trace_view, request)
html = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{title}</title>
<style>
:root {{
color-scheme: light;
--bg: #f5f1e8;
--panel: #fffaf2;
--line: #d8cdbf;
--ink: #2b241d;
--muted: #766758;
--error: #8f2d1f;
--warning: #9a6b00;
--info: #295f8a;
--debug: #5d5d7a;
--link: #7b2f24;
}}
body {{
margin: 0;
background: linear-gradient(180deg, #efe6d8 0%, var(--bg) 100%);
color: var(--ink);
font: 16px/1.5 Georgia, "Times New Roman", serif;
}}
.page {{
max-width: 1200px;
margin: 0 auto;
padding: 24px;
}}
.card {{
background: var(--panel);
border: 1px solid var(--line);
border-radius: 14px;
box-shadow: 0 12px 40px rgba(43, 36, 29, 0.08);
padding: 20px 22px;
margin-bottom: 20px;
}}
h1, h2 {{
margin: 0 0 12px;
font-weight: 700;
}}
h1 {{
font-size: 28px;
}}
h2 {{
font-size: 18px;
color: var(--muted);
text-transform: uppercase;
letter-spacing: 0.06em;
}}
.meta {{
margin: 0;
color: var(--muted);
}}
.trace-links {{
margin: 12px 0 0;
padding-left: 20px;
}}
.trace-links li {{
margin: 6px 0;
}}
a {{
color: var(--link);
text-decoration: none;
}}
a:hover {{
text-decoration: underline;
}}
table {{
width: 100%;
border-collapse: collapse;
}}
th, td {{
vertical-align: top;
padding: 14px 12px;
border-top: 1px solid var(--line);
}}
th {{
text-align: left;
color: var(--muted);
font-size: 13px;
text-transform: uppercase;
letter-spacing: 0.06em;
}}
td.step {{
width: 240px;
font-weight: 700;
white-space: nowrap;
}}
.log-entry + .log-entry {{
margin-top: 14px;
padding-top: 14px;
border-top: 1px dashed var(--line);
}}
.log-meta {{
font-size: 13px;
color: var(--muted);
margin-bottom: 6px;
}}
.message {{
white-space: pre-wrap;
word-break: break-word;
}}
.attrs {{
margin-top: 8px;
padding: 10px 12px;
border-radius: 8px;
background: #f3ebde;
white-space: pre-wrap;
overflow-x: auto;
font-size: 13px;
}}
.level-ERROR {{
color: var(--error);
}}
.level-WARNING {{
color: var(--warning);
}}
.level-INFO {{
color: var(--info);
}}
.level-DEBUG {{
color: var(--debug);
}}
</style>
</head>
<body>
<div class="page">
<section class="card">
<h1>{title}</h1>
<p class="meta">Parent: {escape(trace_view.parent_id or "")}</p>
<p class="meta">Children: {len(trace_view.child_ids)}</p>
</section>
<section class="card">
<h2>Related Traces</h2>
{header_links}
</section>
<section class="card">
<h2>Logs By Step</h2>
<table>
<thead>
<tr><th>Step</th><th>Log</th></tr>
</thead>
<tbody>
{rows}
</tbody>
</table>
</section>
</div>
</body>
</html>"""
return HTMLResponse(content=html)
def _child_id_lines(self, child_ids: tuple[str, ...]) -> list[str]:
lines = ["child_ids:"]
lines.extend(f" - {child_id}" for child_id in child_ids)
return lines
def _text_message(self, record: TraceLogRecord, include_attrs_json: bool) -> str:
if not include_attrs_json:
return record.message
return f"{record.message}, {json.dumps(record.attrs_json, ensure_ascii=False, separators=(',', ':'))}"
def _related_trace_links(self, trace_view: TraceLogView, request: TraceQueryRequest) -> str:
links = [self._trace_link_item("current", trace_view.trace_id, request)]
if trace_view.parent_id:
links.append(self._trace_link_item("parent", trace_view.parent_id, request))
links.extend(self._trace_link_item("child", child_id, request) for child_id in trace_view.child_ids)
return f"<ul class=\"trace-links\">{''.join(links)}</ul>"
def _trace_link_item(self, label: str, trace_id: str, request: TraceQueryRequest) -> str:
href = escape(self._trace_href(trace_id, request), quote=True)
text = escape(f"{label}: {trace_id}")
return f"<li><a href=\"{href}\">{text}</a></li>"
def _trace_href(self, trace_id: str, request: TraceQueryRequest) -> str:
params = urlencode(
{
"format": "html",
"levels": ",".join(request.levels),
"attrs_json": "true" if request.include_attrs_json else "false",
}
)
return f"/traces/{trace_id}?{params}"
def _step_rows(self, trace_view: TraceLogView, request: TraceQueryRequest) -> str:
groups = self._step_groups(trace_view.records)
if not groups:
return "<tr><td class=\"step\">&nbsp;</td><td>No log records for selected filters.</td></tr>"
return "".join(self._step_row(group, request.include_attrs_json) for group in groups)
def _step_groups(self, records: tuple[TraceLogRecord, ...]) -> tuple[TraceStepGroup, ...]:
grouped: dict[str, list[TraceLogRecord]] = {}
order: list[str] = []
for record in records:
step = str(record.step or "")
if step not in grouped:
grouped[step] = []
order.append(step)
grouped[step].append(record)
return tuple(TraceStepGroup(step=step, records=tuple(grouped[step])) for step in order)
def _step_row(self, group: TraceStepGroup, include_attrs_json: bool) -> str:
step_label = escape(group.step) if group.step else "&nbsp;"
entries = "".join(self._log_entry(record, include_attrs_json) for record in group.records)
return f"<tr><td class=\"step\">{step_label}</td><td>{entries}</td></tr>"
def _log_entry(self, record: TraceLogRecord, include_attrs_json: bool) -> str:
level = escape(record.level)
status = escape(record.status or "")
timestamp = escape(record.event_time.isoformat())
message = escape(record.message)
attrs = ""
if include_attrs_json:
attrs_payload = escape(json.dumps(record.attrs_json, ensure_ascii=False, indent=2, sort_keys=True))
attrs = f"<div class=\"attrs\">{attrs_payload}</div>"
return (
f"<article class=\"log-entry level-{level}\">"
f"<div class=\"log-meta\">{timestamp} | {level} | {status}</div>"
f"<div class=\"message\">{message}</div>"
f"{attrs}"
f"</article>"
)
tests/test_trace_endpoint.py
+31
View File
@@ -181,6 +181,37 @@ def test_trace_endpoint_returns_json_payload() -> None:
}
def test_trace_endpoint_returns_html_page_with_related_links() -> None:
async def trace_provider(_trace_id: str, _request: TraceQueryRequest) -> TraceLogView:
return TraceLogView(
trace_id="trace-1",
parent_id="parent-1",
child_ids=("child-1", "child-2"),
records=(
_trace_record(row_id=1, level="INFO", message="loaded prices", step="load_stocks", status="ok"),
_trace_record(row_id=2, level="WARNING", message="filtered suspicious ticker", step="filter_stocks", status="degraded"),
),
)
client = _build_client(trace_provider)
try:
response = client.get("/traces/trace-1?format=html&attrs_json=true")
finally:
client.close()
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/html")
assert "<th>Step</th><th>Log</th>" in response.text
assert ">load_stocks<" in response.text
assert ">filter_stocks<" in response.text
assert "loaded prices" in response.text
assert "filtered suspicious ticker" in response.text
assert 'href="/traces/trace-1?format=html&amp;levels=ERROR%2CWARNING%2CINFO&amp;attrs_json=true"' in response.text
assert 'href="/traces/parent-1?format=html&amp;levels=ERROR%2CWARNING%2CINFO&amp;attrs_json=true"' in response.text
assert 'href="/traces/child-1?format=html&amp;levels=ERROR%2CWARNING%2CINFO&amp;attrs_json=true"' in response.text
assert 'href="/traces/child-2?format=html&amp;levels=ERROR%2CWARNING%2CINFO&amp;attrs_json=true"' in response.text
def test_trace_endpoint_validates_query_params() -> None:
client = _build_client(lambda _trace_id, _request: None)
try: