From 54713a833db9a02fefa91701691b621458c374a9 Mon Sep 17 00:00:00 2001 From: Konrad Rokicki Date: Thu, 17 Sep 2026 14:57:36 -0400 Subject: [PATCH] Log time-to-first-byte, and script the Kibana dashboard The access log measured only total request duration, which for a proxy that streams multi-GB objects is dominated by transfer time: a client pulling 27GB over a slow link was indistinguishable from a stalled backend. Response headers only go out once the upstream store has answered, so the time to http.response.start separates what x2s3 and its backend cost from the body moving at whatever rate the client can take it. The middleware already intercepts that message, so recording it costs one timestamp; it is reported as labels.ttfb_ms in JSON mode (ECS has no time-to-first-byte field) and appended to the text access line. scripts/kibana_dashboard.py probes the log index and builds the production overview dashboard from that measurement set: traffic, latency, throughput, per-target and per-client tables, client classification, and the failing-request stream. It creates saved objects via _bulk_create rather than _import, because _import replays Kibana's pre-8 Lens migrations against hand-written state and returns a bare HTTP 500. Co-Authored-By: Claude Opus 5 (1M context) --- scripts/kibana_dashboard.py | 501 ++++++++++++++++++++++++++++++++++++ tests/test_log.py | 19 ++ x2s3/log.py | 22 +- 3 files changed, 539 insertions(+), 3 deletions(-) create mode 100644 scripts/kibana_dashboard.py diff --git a/scripts/kibana_dashboard.py b/scripts/kibana_dashboard.py new file mode 100644 index 0000000..6ff6725 --- /dev/null +++ b/scripts/kibana_dashboard.py @@ -0,0 +1,501 @@ +#!/usr/bin/env python3 +"""Probe the x2s3 log index in Kibana and (re)build the production dashboard. + +Credentials come from the environment or var/kibana.env, which is where this +project already keeps secrets and is gitignored. Not the repo-root .env: pydantic +reads that one into Settings, which forbids unknown keys and would refuse to +start the app. Either an API key: + + KIBANA_API_KEY= API keys> + +a session cookie copied from the browser (works when Kibana is behind SSO): + + KIBANA_COOKIE= + +or a native-realm username and password: + + KIBANA_USER=you + KIBANA_PASS=... + +Usage: + scripts/kibana_dashboard.py probe # what's actually in the index + scripts/kibana_dashboard.py push # create/overwrite the dashboard + scripts/kibana_dashboard.py dump # print the NDJSON, import nothing +""" +import base64 +import json +import os +import sys +import urllib.error +import urllib.request + +BASE = os.environ.get("KIBANA_URL", "https://kibana-ocp.int.janelia.org").rstrip("/") +SPACE = os.environ.get("KIBANA_SPACE", "s3proxy") +DV_TITLE = os.environ.get("KIBANA_DATA_VIEW", "filebeat-s3proxy-*") + +# Logical name -> field in the index. Defaults are the ECS names x2s3/log.py +# writes; probe reports what the shipper actually indexed, and a shipper that +# nests or keyword-suffixes them only needs fixing here. +F = { + "status": "http.response.status_code", + "duration": "event.duration_ms", + "bytes": "http.response.body.bytes", + "method": "http.request.method", + "target": "labels.target", + "path": "url.path", + "client": "client.ip", + "agent": "user_agent.original", + "level": "log.level", + "trace": "trace.id", + "host": "host.name", + "referrer": "http.request.referrer", +} + +# Client classification. Matched against the analyzed .text subfield so the terms +# are case-insensitive and match anywhere in the UA string. +# +# These two groups are the ones a user agent can actually identify. The bulk of +# the non-human traffic is a distributed scraper that sends byte-identical real +# Chrome strings from thousands of addresses, so no UA rule can catch it; the +# "user agents by unique clients" panel is what exposes it, and blocking it needs +# a network-level rule x2s3's log has no field for. +MONITORING = ["bot", "crawler", "spider", "uptime", "monitoring", + "gatus", "pingdom", "zabbix", "nagios", "healthcheck"] +TOOLS = ["aws", "boto", "botocore", "ahc", "curl", "wget", "okhttp", "aiohttp", + "python", "java", "s5cmd", "rclone", "minio", "go-http-client"] +# Every access-log line has a status code; startup and stream-progress lines do +# not. Filtering on it keeps counts honest without depending on event.dataset +# surviving the shipper. +DASHBOARD_QUERY = f"{F['status']}: *" + + +def load_dotenv(path): + if not os.path.exists(path): + return + for line in open(path): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip().strip("'\"")) + + +def auth_headers(): + """Whichever of the three credential styles var/kibana.env supplies. + + An SSO-backed stack has no native-realm password to send, so an API key or a + copied browser session cookie are the only ones that authenticate there. + """ + if os.environ.get("KIBANA_API_KEY"): + return {"Authorization": f"ApiKey {os.environ['KIBANA_API_KEY']}"} + if os.environ.get("KIBANA_COOKIE"): + cookie = os.environ["KIBANA_COOKIE"] + return {"Cookie": cookie if "=" in cookie else f"sid={cookie}"} + user, password = os.environ.get("KIBANA_USER"), os.environ.get("KIBANA_PASS") + if not (user and password): + sys.exit("Set KIBANA_API_KEY, KIBANA_COOKIE, or KIBANA_USER/KIBANA_PASS " + "in var/kibana.env") + return {"Authorization": "Basic " + + base64.b64encode(f"{user}:{password}".encode()).decode()} + + +def call(path, data=None, headers=None, method=None, missing_ok=False): + url = f"{BASE}/s/{SPACE}{path}" + request = urllib.request.Request(url, data=data, method=method or ("POST" if data else "GET")) + for key, value in auth_headers().items(): + request.add_header(key, value) + request.add_header("kbn-xsrf", "true") + for key, value in (headers or {}).items(): + request.add_header(key, value) + try: + with urllib.request.urlopen(request, timeout=120) as response: + return json.load(response) + except urllib.error.HTTPError as error: + if missing_ok and error.code == 404: + return None + body = error.read().decode(errors="replace")[:2000] + sys.exit(f"{error.code} {error.reason} from {path}\n{body}") + + +def es(path, query=None): + """Run an Elasticsearch request through Kibana's console proxy.""" + # The proxy route is itself POST-only; the ES verb goes in &method=. + return call( + f"/api/console/proxy?path={urllib.request.quote(path, safe='/*_,')}" + f"&method={'POST' if query else 'GET'}", + data=json.dumps(query).encode() if query is not None else b"", + headers={"Content-Type": "application/json"}, + method="POST", + ) + + +def kibana_major(): + if os.environ.get("KIBANA_VERSION"): + return int(os.environ["KIBANA_VERSION"].split(".")[0]) + # This deployment's /api/status omits the version block, so assume the + # current major and let KIBANA_VERSION override if the import complains. + status = call("/api/status") + return int(status.get("version", {}).get("number", "8.0.0").split(".")[0]) + + +def data_view_id(): + for view in call("/api/data_views")["data_view"]: + if view["title"] == DV_TITLE: + return view["id"] + sys.exit(f"No data view titled {DV_TITLE!r}; found: " + + ", ".join(v["title"] for v in call("/api/data_views")["data_view"])) + + +# --------------------------------------------------------------------------- probe + +def probe(): + print(f"kibana major: {kibana_major()}") + print(f"data view : {DV_TITLE} -> {data_view_id()}") + + caps = es(f"{DV_TITLE}/_field_caps?fields=*")["fields"] + print("\nfields (expected -> indexed type):") + for name, field in sorted(F.items()): + types = sorted(caps.get(field, {})) + keyword = f"{field}.keyword" in caps + note = " (.keyword exists)" if keyword else "" + print(f" {name:9} {field:32} {','.join(types) or 'MISSING'}{note}") + extra = [f for f in caps if f.startswith(("labels.", "service.", "event.", "http.", "url."))] + print("\nother log fields present: " + ", ".join(sorted(extra)[:40])) + + result = es(f"{DV_TITLE}/_search", { + "size": 1, + "sort": [{"@timestamp": "desc"}], + "query": {"bool": {"filter": [ + {"exists": {"field": F["status"]}}, + {"range": {"@timestamp": {"gte": "now-24h"}}}, + ]}}, + "aggs": { + "status": {"terms": {"field": F["status"], "size": 10}}, + "targets": {"terms": {"field": F["target"], "size": 10}}, + "levels": {"terms": {"field": F["level"], "size": 5}}, + "p95": {"percentiles": {"field": F["duration"], "percents": [50, 95, 99]}}, + "bytes": {"sum": {"field": F["bytes"]}}, + }, + }) + total = result["hits"]["total"] + print(f"\naccess lines in last 24h: {total.get('value')} ({total.get('relation')})") + hits = result["hits"]["hits"] + if hits: + print("newest document:\n" + json.dumps(hits[0]["_source"], indent=2)[:2500]) + for name in ("status", "targets", "levels"): + buckets = result["aggregations"][name]["buckets"] + print(f"{name}: " + ", ".join(f"{b['key']}={b['doc_count']}" for b in buckets)) + print("latency ms: " + json.dumps(result["aggregations"]["p95"]["values"])) + print(f"bytes served 24h: {result['aggregations']['bytes']['value']:,.0f}") + + +# ------------------------------------------------------------------- dashboard build + +def columns(major): + records = "___records___" if major >= 8 else "Records" + + def count(label="Requests"): + return {"label": label, "dataType": "number", "operationType": "count", + "isBucketed": False, "scale": "ratio", "sourceField": records} + + def count_where(label, query): + column = count(label) + column["filter"] = {"query": query, "language": "kuery"} + return column + + def unique(field, label): + return {"label": label, "dataType": "number", "operationType": "unique_count", + "sourceField": field, "isBucketed": False, "scale": "ratio"} + + def percentile(p, label=None): + return {"label": label or f"p{p} latency", "dataType": "number", + "operationType": "percentile", "sourceField": F["duration"], + "isBucketed": False, "scale": "ratio", + "params": {"percentile": p, "format": {"id": "number", "params": { + "decimals": 1, "suffix": " ms"}}}} + + def total_bytes(label="Bytes served"): + return {"label": label, "dataType": "number", "operationType": "sum", + "sourceField": F["bytes"], "isBucketed": False, "scale": "ratio", + "params": {"format": {"id": "bytes", "params": {"decimals": 1}}}} + + def date_histogram(): + return {"label": "@timestamp", "dataType": "date", "operationType": "date_histogram", + "sourceField": "@timestamp", "isBucketed": True, "scale": "interval", + "params": {"interval": "auto", "includeEmptyRows": True, "dropPartials": False}} + + def terms(field, size, order_by, label): + return {"label": label, "dataType": "string", "operationType": "terms", + "sourceField": field, "isBucketed": True, "scale": "ordinal", + "params": {"size": size, "orderBy": {"type": "column", "columnId": order_by}, + "orderDirection": "desc", "otherBucket": False, + "missingBucket": False, "parentFormat": {"id": "terms"}}} + + def client_classes(): + ua = f"{F['agent']}.text" + mon = f"{ua}: ({' or '.join(MONITORING)})" + tool = f"{ua}: ({' or '.join(TOOLS)})" + browser = f"{ua}: mozilla" + referred = f"{F['referrer']}: *" + # Mutually exclusive, so the stack sums to total traffic. + rest = f"not {mon} and not {tool}" + classes = [ + ("Monitoring", mon), + ("SDK / tool", f"not {mon} and {tool}"), + ("Browser, referred", f"{rest} and {browser} and {referred}"), + ("Browser, no referrer", f"{rest} and {browser} and not {referred}"), + ("Other", f"{rest} and not {browser}"), + ] + return {"label": "Client type", "dataType": "string", "operationType": "filters", + "isBucketed": True, "scale": "ordinal", + "params": {"filters": [{"label": label, + "input": {"language": "kuery", "query": query}} + for label, query in classes]}} + + def status_classes(): + ranges = [("2xx success", 200, 300), ("3xx redirect", 300, 400), + ("4xx client error", 400, 500), ("5xx server error", 500, 600)] + return {"label": "Status", "dataType": "string", "operationType": "filters", + "isBucketed": True, "scale": "ordinal", + "params": {"filters": [ + {"label": label, "input": {"language": "kuery", "query": + f"{F['status']} >= {low} and {F['status']} < {high}"}} + for label, low, high in ranges]}} + + return (count, count_where, unique, percentile, total_bytes, date_histogram, + terms, status_classes, client_classes) + + +def lens(obj_id, title, vis_type, visualization, cols, order, dv, major, query=""): + ds_key = "formBased" if major >= 8 else "indexpattern" + refs = [{"type": "index-pattern", "id": dv, "name": "indexpattern-datasource-layer-l1"}] + if major < 8: + refs.append({"type": "index-pattern", "id": dv, + "name": "indexpattern-datasource-current-indexpattern"}) + return {"id": obj_id, "type": "lens", "references": refs, "attributes": { + "title": title, "description": "", "visualizationType": vis_type, + "state": { + "visualization": visualization, + "query": {"query": query, "language": "kuery"}, + "filters": [], + "datasourceStates": {ds_key: {"layers": {"l1": { + "columnOrder": order, "columns": cols, "incompleteColumns": {}}}}}, + }}} + + +def xy(series_type, x, y, split=None, legend="bottom"): + """legend=None for a single-series chart the title already names. + + Legends default to the bottom edge rather than the right. A right-hand legend + is laid out inside the panel and takes 130-300px off the plot, so two stacked + charts whose legends differ in width get plot areas of different widths and + their time axes no longer line up. A bottom legend costs height instead, which + leaves every plot spanning the full panel width and the x-axes aligned. + """ + layer = {"layerId": "l1", "layerType": "data", "seriesType": series_type, + "xAccessor": x, "accessors": y} + if split: + layer["splitAccessor"] = split + # 'position' only accepts an edge; hiding a legend is isVisible, not a position. + return {"preferredSeriesType": series_type, + "legend": {"isVisible": legend is not None, "position": legend or "right"}, + "valueLabels": "hide", "fittingFunction": "None", + "axisTitlesVisibilitySettings": {"x": False, "yLeft": False, "yRight": False}, + "gridlinesVisibilitySettings": {"x": False, "yLeft": True, "yRight": False}, + "tickLabelsVisibilitySettings": {"x": True, "yLeft": True, "yRight": False}, + "layers": [layer]} + + +def metric_vis(accessor, major): + if major >= 8: + return {"layerId": "l1", "layerType": "data", "metricAccessor": accessor} + return {"layerId": "l1", "layerType": "data", "accessor": accessor} + + +def exclude_filter(alias, terms, dv, index): + """A negated, switched-off filter pill matching any of `terms` in the UA. + + Shipped disabled so the panels still show the full picture; enabling one is a + single click in the filter bar, which is the only honest way to offer this + given that the terms are a heuristic. + """ + return { + "meta": {"alias": alias, "negate": True, "disabled": True, "type": "custom", + "key": "query", "index": dv}, + "query": {"bool": {"minimum_should_match": 1, "should": [ + {"match": {f"{F['agent']}.text": term}} for term in terms]}}, + "$state": {"store": "appState"}, + } + + +def build(dv, major): + (count, count_where, unique, percentile, total_bytes, date_histogram, + terms, status_classes, client_classes) = columns(major) + objects = [] + + def add(obj_id, title, vis_type, vis, cols, order, grid, query=""): + objects.append((lens(obj_id, title, vis_type, vis, cols, order, dv, major, query), grid)) + + # Grid slots are (x, y, w, h) on Kibana's 48-column grid, and mirror the + # arrangement as hand-tuned in the UI: a 3x2 block of headline numbers, then + # the four comparable full-width time series stacked so their x-axes line up, + # then each remaining series beside the table that breaks it down. + + # Volume, row 1. + add("x2s3-requests", "Requests", "lnsMetric", metric_vis("m", major), + {"m": count()}, ["m"], (0, 0, 10, 7)) + add("x2s3-bytes", "Bytes served", "lnsMetric", metric_vis("m", major), + {"m": total_bytes()}, ["m"], (10, 0, 10, 7)) + add("x2s3-unique-clients", "Unique clients", "lnsMetric", metric_vis("m", major), + {"m": unique(F["client"], "Unique IPs")}, ["m"], (20, 0, 10, 7)) + + add("x2s3-top-targets", "Top targets", "lnsXY", + xy("bar_horizontal", "b", ["m"], legend=None), + {"b": terms(F["target"], 10, "m", "Target"), "m": count()}, ["b", "m"], + (30, 0, 18, 14)) + + # Health, row 2, sitting under the volume number each one qualifies. + add("x2s3-p95", "p95 latency", "lnsMetric", metric_vis("m", major), + {"m": percentile(95)}, ["m"], (0, 7, 10, 7)) + add("x2s3-server-errors", "Server errors (5xx)", "lnsMetric", metric_vis("m", major), + {"m": count("5xx")}, ["m"], (10, 7, 10, 7), query=f"{F['status']} >= 500") + # 499 is nginx's client-closed-connection: a viewer that gave up mid-stream + # rather than anything x2s3 did wrong, so it is worth its own number. + add("x2s3-client-aborts", "Client aborts (499)", "lnsMetric", metric_vis("m", major), + {"m": count("499")}, ["m"], (20, 7, 10, 7), query=f"{F['status']}: 499") + + # Three stacked full-width rows, meant to be read against each other. + add("x2s3-rate-by-status", "Request rate by status class", "lnsXY", + xy("bar_stacked", "t", ["m"], split="s"), + {"t": date_histogram(), "s": status_classes(), "m": count()}, ["t", "s", "m"], + (0, 14, 48, 12)) + add("x2s3-by-node", "Requests by node", "lnsXY", + xy("bar_stacked", "t", ["m"], split="h"), + {"t": date_histogram(), "h": terms(F["host"], 5, "m", "Node"), "m": count()}, + ["t", "h", "m"], (0, 26, 48, 12)) + add("x2s3-unique-clients-over-time", "Unique clients over time", "lnsXY", + xy("line", "t", ["m"], legend=None), + {"t": date_histogram(), "m": unique(F["client"], "Unique IPs")}, ["t", "m"], + (0, 38, 48, 12)) + + add("x2s3-by-client-type", "Traffic by client type", "lnsXY", + xy("bar_stacked", "t", ["m"], split="c"), + {"t": date_histogram(), "c": client_classes(), "m": count()}, ["t", "c", "m"], + (0, 50, 48, 12)) + + # Latency and throughput. + add("x2s3-latency-percentiles", "Latency percentiles", "lnsXY", + xy("line", "t", ["p50", "p95", "p99"]), + {"t": date_histogram(), "p50": percentile(50, "p50"), + "p95": percentile(95, "p95"), "p99": percentile(99, "p99")}, + ["t", "p50", "p95", "p99"], (0, 62, 48, 13)) + add("x2s3-throughput", "Throughput", "lnsXY", + xy("area", "t", ["m"], legend=None), + {"t": date_histogram(), "m": total_bytes("Bytes")}, ["t", "m"], (0, 75, 48, 11)) + + # Per-target and per-client detail. + add("x2s3-target-table", "Targets: traffic and latency", "lnsDatatable", + {"layerId": "l1", "layerType": "data", + "columns": [{"columnId": c} for c in ("b", "m", "p95", "bytes")]}, + {"b": terms(F["target"], 20, "m", "Target"), "m": count(), + "p95": percentile(95), "bytes": total_bytes()}, ["b", "m", "p95", "bytes"], + (0, 86, 25, 14)) + add("x2s3-client-table", "Top clients", "lnsDatatable", + {"layerId": "l1", "layerType": "data", + "columns": [{"columnId": c} for c in ("b", "m", "bytes", "errors")]}, + {"b": terms(F["client"], 20, "m", "Client IP"), "m": count(), + "bytes": total_bytes(), + "errors": count_where("Errors", f"{F['status']} >= 400")}, + ["b", "m", "bytes", "errors"], (25, 86, 23, 14)) + + add("x2s3-user-agents", "User agents by unique clients", "lnsDatatable", + {"layerId": "l1", "layerType": "data", + "columns": [{"columnId": c} for c in ("b", "ips", "m", "bytes", "errors")]}, + {"b": terms(F["agent"], 25, "ips", "User agent"), + "ips": unique(F["client"], "Unique IPs"), "m": count(), + "bytes": total_bytes(), + "errors": count_where("Errors", f"{F['status']} >= 400")}, + ["b", "ips", "m", "bytes", "errors"], (0, 100, 48, 16)) + + # The actual failing requests. + objects.append(({"id": "x2s3-errors-search", "type": "search", "attributes": { + "title": "Failed requests (4xx/5xx)", + "description": "", + "columns": [F["status"], F["method"], F["path"], F["target"], F["client"], F["trace"]], + "sort": [["@timestamp", "desc"]], + "kibanaSavedObjectMeta": {"searchSourceJSON": json.dumps({ + "query": {"query": f"{F['status']} >= 400", "language": "kuery"}, + "filter": [], + "indexRefName": "kibanaSavedObjectMeta.searchSourceJSON.index"})}, + }, "references": [{"type": "index-pattern", "id": dv, + "name": "kibanaSavedObjectMeta.searchSourceJSON.index"}]}, + (0, 116, 48, 18))) + + panels, references = [], [] + for index, (obj, (x, y, w, h)) in enumerate(objects, start=1): + panels.append({"panelIndex": str(index), "type": obj["type"], + "gridData": {"x": x, "y": y, "w": w, "h": h, "i": str(index)}, + "embeddableConfig": {"enhancements": {}}, + "panelRefName": f"panel_{index}"}) + references.append({"name": f"{index}:panel_{index}", "type": obj["type"], "id": obj["id"]}) + + dashboard = {"id": "x2s3-overview", "type": "dashboard", "references": references, + "attributes": { + "title": "x2s3 — Production Overview", + "description": "Traffic, latency, throughput and errors from the x2s3 access log.", + "panelsJSON": json.dumps(panels), + "optionsJSON": json.dumps({"hidePanelTitles": False, "useMargins": True, + "syncColors": False, "syncCursor": True, + "syncTooltips": True}), + "timeRestore": True, "timeFrom": "now-24h", "timeTo": "now", + "refreshInterval": {"pause": False, "value": 60000}, + "kibanaSavedObjectMeta": {"searchSourceJSON": json.dumps({ + "query": {"query": DASHBOARD_QUERY, "language": "kuery"}, + "filter": [exclude_filter("Exclude monitoring bots", MONITORING, dv, 0), + exclude_filter("Exclude SDK & tool clients", TOOLS, dv, 1)]})}, + "version": 1}} + return "\n".join(json.dumps(obj) for obj, _ in objects + [(dashboard, None)]) + + +def push(ndjson): + """Create the objects directly rather than through _import. + + _import runs the type's migration chain, and a hand-written Lens object with + no migration version makes Kibana replay its pre-8 migrations and 500. + _bulk_create stamps the running version instead, so the state is taken as-is. + """ + objects = [json.loads(line) for line in ndjson.splitlines()] + + # The time range and refresh interval are dashboard settings someone may have + # changed in the UI; keep whatever is there rather than resetting it. + existing = call("/api/saved_objects/dashboard/x2s3-overview", missing_ok=True) + if existing: + kept = {key: existing["attributes"][key] + for key in ("timeFrom", "timeTo", "refreshInterval") + if key in existing["attributes"]} + for obj in objects: + if obj["type"] == "dashboard": + obj["attributes"].update(kept) + print(f" keeping time range {kept.get('timeFrom')} -> {kept.get('timeTo')}") + result = call("/api/saved_objects/_bulk_create?overwrite=true", + data=json.dumps(objects).encode(), + headers={"Content-Type": "application/json"}) + failed = [o for o in result["saved_objects"] if o.get("error")] + for obj in result["saved_objects"]: + print(f" {obj['type']:10} {obj['id']:26} " + f"{json.dumps(obj['error']) if obj.get('error') else 'ok'}") + if failed: + sys.exit(f"{len(failed)} object(s) failed") + print(f"\n{BASE}/s/{SPACE}/app/dashboards#/view/x2s3-overview") + + +if __name__ == "__main__": + load_dotenv(os.path.join(os.path.dirname(__file__), os.pardir, "var", "kibana.env")) + command = sys.argv[1] if len(sys.argv) > 1 else "probe" + if command == "probe": + probe() + elif command == "dump": + print(build(data_view_id(), kibana_major())) + elif command == "push": + push(build(data_view_id(), kibana_major())) + else: + sys.exit(__doc__) diff --git a/tests/test_log.py b/tests/test_log.py index 61a4144..69e8748 100644 --- a/tests/test_log.py +++ b/tests/test_log.py @@ -79,3 +79,22 @@ def test_text_mode_stays_human_readable(capsys): err = capsys.readouterr().err assert '"GET /local-files/README.md HTTP/1.1" 200' in err assert "target=local-files" in err + + +def test_json_mode_splits_ttfb_from_total_duration(capsys): + """TTFB must be present and no larger than the total, or the split is a lie.""" + with _client("json") as client: + client.get("/local-files/README.md") + + record = _access_lines(capsys.readouterr())[0] + ttfb_ms = record["labels.ttfb_ms"] + assert ttfb_ms > 0 + # event.duration is nanoseconds; TTFB is part of it, not additional to it. + assert ttfb_ms <= record["event.duration"] / 1_000_000 + 0.001 + + +def test_text_mode_includes_ttfb(capsys): + with _client("text") as client: + client.get("/local-files/README.md") + + assert "ttfb=" in capsys.readouterr().err diff --git a/x2s3/log.py b/x2s3/log.py index 9388900..5c5607b 100644 --- a/x2s3/log.py +++ b/x2s3/log.py @@ -136,6 +136,12 @@ class AccessLogMiddleware: sent, so the duration measured here is the real transfer time rather than time-to-first-byte, and the response bytes can be counted on the way past. + Both halves of that are logged separately. The response headers only go out + once the upstream store has answered, so the time to ``http.response.start`` + is what x2s3 and its backend cost, and the rest of the duration is the body + moving at whatever rate the client can take it. Without the split, one client + pulling 27GB over a slow link is indistinguishable from a stalled backend. + Registered outermost so it also covers responses generated by the CORS and PNA middlewares. It reads the request id and target name that the inner layers leave in ``scope["state"]``, which is the same dict object all the @@ -153,11 +159,13 @@ async def __call__(self, scope, receive, send): start = time.perf_counter() status = 500 body_bytes = 0 + ttfb_s = None async def send_with_accounting(message): - nonlocal status, body_bytes + nonlocal status, body_bytes, ttfb_s if message["type"] == "http.response.start": status = message["status"] + ttfb_s = time.perf_counter() - start elif message["type"] == "http.response.body": body_bytes += len(message.get("body", b"")) await send(message) @@ -165,10 +173,11 @@ async def send_with_accounting(message): try: await self.app(scope, receive, send_with_accounting) finally: - _log_request(scope, status, body_bytes, time.perf_counter() - start) + _log_request(scope, status, body_bytes, time.perf_counter() - start, ttfb_s) -def _log_request(scope, status: int, body_bytes: int, duration_s: float): +def _log_request(scope, status: int, body_bytes: int, duration_s: float, + ttfb_s: float | None): client = scope.get("client") or ("unknown", 0) http_version = scope.get("http_version", "1.1") @@ -195,6 +204,8 @@ def _log_request(scope, status: int, body_bytes: int, duration_s: float): f'HTTP/{http_version}" ' f"{status} {body_bytes}b {duration_s * 1000:.2f}ms" ) + if ttfb_s is not None: + message += f" ttfb={ttfb_s * 1000:.2f}ms" if target: message += f" target={target}" @@ -210,6 +221,11 @@ def _log_request(scope, status: int, body_bytes: int, duration_s: float): "client.ip": client[0], "client.port": client[1], } + if ttfb_s is not None: + # ECS has no time-to-first-byte field, so it goes under labels like the + # other custom dimensions. Milliseconds, because event.duration's + # nanoseconds only exist to satisfy the ECS definition. + fields["labels.ttfb_ms"] = round(ttfb_s * 1000, 3) if request_id: fields["trace.id"] = request_id if query: