rays
ccc4766b03
`query.py flame '<TraceQL>' [since] [out.svg]` folds the matching spans of up to 100 traces into stacks of self time and draws them as a flame graph, with the folded stacks beside it for speedscope or inferno. A wall-clock picture of where a route's time goes, waiting included, which a CPU profiler would not show: ipx waits on Postgres and the disk far more than it computes. The traces are fetched in one container, a docker run apiece being a second each. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
231 lines
12 KiB
Python
Executable File
231 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Ask production's Loki, Tempo or Prometheus a question, from anywhere that can run docker on Tower.
|
|
|
|
query.py logs '<LogQL log query>' [since] lines, newest first
|
|
query.py metric '<LogQL metric query>' [since] one value per series; $range is `since`
|
|
query.py traces '<TraceQL query>' [since] matching traces, slowest first
|
|
query.py spans '<TraceQL query>' [since] the matching spans, each with its own time
|
|
query.py trace <trace id> one trace's spans, as a tree
|
|
query.py prom '<PromQL query>' [since] one value per series, now; $range is `since`
|
|
query.py flame '<TraceQL query>' [since] [out.svg] where the matching spans' time
|
|
went, summed over up to 100 traces, as a
|
|
flame graph (and out.svg.folded)
|
|
|
|
`since` is 1h, 24h, 7d and the like (default 24h). Loki and Tempo publish no query port on the
|
|
host, so each call runs a throwaway alpine container on the monitoring project's network.
|
|
"""
|
|
import base64, html, json, subprocess, sys, time, urllib.parse, zlib
|
|
|
|
NET = "monitoring_default"
|
|
|
|
|
|
def fetch(url):
|
|
out = subprocess.run(["docker", "run", "--rm", "--network", NET, "alpine", "wget", "-qO-", url],
|
|
capture_output=True, text=True, timeout=120)
|
|
if out.returncode:
|
|
sys.exit(f"query failed: {out.stderr.strip() or out.stdout.strip()}\n{url}")
|
|
return json.loads(out.stdout)
|
|
|
|
|
|
def fetch_many(urls):
|
|
# One container for them all: a docker run apiece is about a second, a hundred traces a minute and a half.
|
|
out = subprocess.run(["docker", "run", "--rm", "--network", NET, "alpine", "sh", "-c",
|
|
'for u; do wget -qO- "$u"; echo; echo @@; done', "--", *urls],
|
|
capture_output=True, text=True, timeout=600)
|
|
return [json.loads(part) for part in out.stdout.split("\n@@\n") if part.strip()]
|
|
|
|
|
|
def trace_spans(r):
|
|
return [sp for b in r.get("batches", r.get("resourceSpans", []))
|
|
for ss in b.get("scopeSpans", b.get("instrumentationLibrarySpans", [])) for sp in ss["spans"]]
|
|
|
|
|
|
def span_hex(sid):
|
|
# A trace's span ids are base64 and the search's are hex.
|
|
return sid if len(sid) == 16 and all(c in "0123456789abcdef" for c in sid) else base64.b64decode(sid).hex()
|
|
|
|
|
|
def label(sp):
|
|
attrs = {a["key"]: next(iter(a["value"].values()), None) for a in sp.get("attributes", [])}
|
|
return f"step:{attrs['of']}" if sp["name"] == "step" and attrs.get("of") else sp["name"]
|
|
|
|
|
|
def fold(spans, roots):
|
|
"""Self time in µs for each stack under the spans in `roots` (hex ids), identical stacks summed.
|
|
|
|
shortcut: only as fine as the spans; a route's time outside its `step`s is its own frame. Add
|
|
steps in ipx where that is large. Children run at once (try_join) can add up past their
|
|
parent, so a frame's self time is floored at 0.
|
|
"""
|
|
kids, stacks = {}, {}
|
|
for sp in spans:
|
|
kids.setdefault(span_hex(sp.get("parentSpanId") or "0" * 16), []).append(sp)
|
|
|
|
def walk(sp, path):
|
|
path = f"{path};{label(sp)}" if path else label(sp)
|
|
dur = lambda s: (int(s["endTimeUnixNano"]) - int(s["startTimeUnixNano"])) // 1000
|
|
cs = kids.get(span_hex(sp["spanId"]), [])
|
|
stacks[path] = stacks.get(path, 0) + max(0, dur(sp) - sum(dur(c) for c in cs))
|
|
for c in cs:
|
|
walk(c, path)
|
|
|
|
for sp in spans:
|
|
if span_hex(sp["spanId"]) in roots:
|
|
walk(sp, "")
|
|
return stacks
|
|
|
|
|
|
def check_fold():
|
|
def sp(sid, parent, name, start, end, of=None):
|
|
attrs = [{"key": "of", "value": {"stringValue": of}}] if of else []
|
|
return {"spanId": sid, "parentSpanId": parent, "name": name, "attributes": attrs,
|
|
"startTimeUnixNano": str(start * 1000), "endTimeUnixNano": str(end * 1000)}
|
|
a, b, c, d = (f"{i:016x}" for i in range(1, 5))
|
|
spans = [sp(a, "", "GET /x", 0, 100), sp(b, a, "step", 0, 60, "q"), sp(c, a, "step", 0, 60, "q"),
|
|
sp(d, b, "inner", 0, 10)]
|
|
assert fold(spans, {a}) == {"GET /x": 0, "GET /x;step:q": 110, "GET /x;step:q;inner": 10}, fold(spans, {a})
|
|
assert fold(spans, {b}) == {"step:q": 50, "step:q;inner": 10}
|
|
# base64 ids, as a trace sends them, match the search's hex.
|
|
assert span_hex(base64.b64encode(bytes.fromhex(a)).decode()) == a
|
|
print("fold: ok")
|
|
|
|
|
|
def flame_svg(stacks, n, title):
|
|
tree = {} # name -> [total, children]
|
|
for path, us in stacks.items():
|
|
node = tree
|
|
for part in path.split(";"):
|
|
node = node.setdefault(part, [0, {}])
|
|
node[0] += us
|
|
node = node[1]
|
|
total = sum(v[0] for v in tree.values()) or 1
|
|
width, row, rects = 1200, 18, []
|
|
|
|
def draw(node, x, depth, trail):
|
|
for name, (us, sub) in sorted(node.items()):
|
|
w = us / total * width
|
|
stack = f"{trail};{name}" if trail else name
|
|
hue = 10 + zlib.crc32(name.encode()) % 45
|
|
tip = f"{stack}\n{us / 1000:.0f} ms in all, {100 * us / total:.1f}%, {us / 1000 / n:.1f} ms a request"
|
|
text = html.escape(name[: int(w / 7)]) if w > 30 else ""
|
|
rects.append(f'<g><title>{html.escape(tip)}</title><rect x="{x:.1f}" y="{30 + depth * row}" '
|
|
f'width="{max(w - 0.5, 0.1):.1f}" height="{row - 1}" fill="hsl({hue},85%,62%)"/>'
|
|
f'<text x="{x + 3:.1f}" y="{30 + depth * row + 13}">{text}</text></g>')
|
|
draw(sub, x, depth + 1, stack)
|
|
x += w
|
|
|
|
draw(tree, 0, 0, "")
|
|
depth = max(p.count(";") for p in stacks) + 1 if stacks else 1
|
|
head = html.escape(f"{title}: {n} request(s), {total / 1000 / n:.1f} ms each on average")
|
|
return (f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{40 + depth * row}" '
|
|
f'font-family="monospace" font-size="11"><rect width="100%" height="100%" fill="#fff"/>'
|
|
f'<text x="4" y="18" font-size="13">{head}</text>{"".join(rects)}</svg>')
|
|
|
|
|
|
def seconds(since):
|
|
return int(since[:-1]) * {"m": 60, "h": 3600, "d": 86400}[since[-1]]
|
|
|
|
|
|
def main():
|
|
if sys.argv[1:] == ["flame", "--check"]:
|
|
return check_fold()
|
|
if len(sys.argv) < 3:
|
|
sys.exit(__doc__)
|
|
kind, q = sys.argv[1], sys.argv[2]
|
|
since = sys.argv[3] if len(sys.argv) > 3 else "24h"
|
|
now = time.time()
|
|
start = now - seconds(since)
|
|
enc = urllib.parse.quote
|
|
if kind == "logs":
|
|
r = fetch(f"http://loki:3100/loki/api/v1/query_range?query={enc(q)}&limit=500"
|
|
f"&start={int(start * 1e9)}&end={int(now * 1e9)}&direction=backward")
|
|
lines = [(ts, line) for s in r["data"]["result"] for ts, line in s["values"]]
|
|
for ts, line in sorted(lines, reverse=True):
|
|
print(line)
|
|
print(f"-- {len(lines)} line(s){' (limit reached)' if len(lines) >= 500 else ''}", file=sys.stderr)
|
|
elif kind == "metric":
|
|
r = fetch(f"http://loki:3100/loki/api/v1/query?query={enc(q.replace('$range', since))}&time={int(now * 1e9)}")
|
|
rows = sorted(r["data"]["result"], key=lambda s: -float(s["value"][1]))
|
|
for s in rows:
|
|
labels = {k: v for k, v in s["metric"].items() if k not in ("container", "compose_project", "service_name")}
|
|
print(f"{s['value'][1]:>12} {json.dumps(labels) if labels else ''}")
|
|
print(f"-- {len(rows)} series", file=sys.stderr)
|
|
elif kind == "traces":
|
|
r = fetch(f"http://tempo:3200/api/search?q={enc(q)}&start={int(start)}&end={int(now)}&limit=100")
|
|
traces = sorted(r.get("traces", []), key=lambda t: -t.get("durationMs", 0))
|
|
for t in traces:
|
|
when = time.strftime("%m-%d %H:%M:%S", time.gmtime(int(t["startTimeUnixNano"]) / 1e9))
|
|
# No root name: its root span is a tap in the page, which the browser never sends.
|
|
name = t.get("rootTraceName") or "(from the page)"
|
|
print(f"{t.get('durationMs', 0):>8}ms {when}Z {t['traceID']} {name}")
|
|
print(f"-- {len(traces)} trace(s)", file=sys.stderr)
|
|
elif kind == "spans":
|
|
# The spans that matched, each with its own time: a trace's length says nothing about a
|
|
# slow request in it when the trace also holds a page's live updates or an audio stream.
|
|
r = fetch(f"http://tempo:3200/api/search?q={enc(q)}&start={int(start)}&end={int(now)}&limit=200&spss=20")
|
|
rows = []
|
|
for t in r.get("traces", []):
|
|
for ss in t.get("spanSets") or [t.get("spanSet") or {}]:
|
|
for sp in ss.get("spans", []):
|
|
attrs = {a["key"]: next(iter(a["value"].values()), None) for a in sp.get("attributes", [])}
|
|
what = attrs.get("http.route") or attrs.get("of") or attrs.get("url.path") or sp.get("name") or ""
|
|
rows.append((int(sp.get("durationNanos", 0)) / 1e6, int(sp["startTimeUnixNano"]), t["traceID"], what))
|
|
for ms, ts, tid, what in sorted(rows, reverse=True):
|
|
when = time.strftime("%m-%d %H:%M:%S", time.gmtime(ts / 1e9))
|
|
print(f"{ms:>8.0f}ms {when}Z {tid} {what}")
|
|
print(f"-- {len(rows)} span(s)", file=sys.stderr)
|
|
elif kind == "prom":
|
|
# ipx's /metrics (#194), scraped as job ipx; through the network like the others.
|
|
r = fetch(f"http://prometheus:9090/api/v1/query?query={enc(q.replace('$range', since))}&time={now}")
|
|
rows = sorted(r["data"]["result"], key=lambda s: -float(s["value"][1]))
|
|
for s in rows:
|
|
labels = {k: v for k, v in s["metric"].items() if k not in ("job", "instance")}
|
|
print(f"{float(s['value'][1]):>14g} {json.dumps(labels) if labels else ''}")
|
|
print(f"-- {len(rows)} series", file=sys.stderr)
|
|
elif kind == "flame":
|
|
out = sys.argv[4] if len(sys.argv) > 4 else "flame.svg"
|
|
r = fetch(f"http://tempo:3200/api/search?q={enc(q)}&start={int(start)}&end={int(now)}&limit=100&spss=20")
|
|
found = r.get("traces", [])
|
|
roots = {sp["spanID"] for t in found for ss in t.get("spanSets") or [t.get("spanSet") or {}]
|
|
for sp in ss.get("spans", [])}
|
|
stacks = {}
|
|
for tr in fetch_many([f"http://tempo:3200/api/traces/{t['traceID']}" for t in found]):
|
|
for path, us in fold(trace_spans(tr), roots).items():
|
|
stacks[path] = stacks.get(path, 0) + us
|
|
n = len(roots) or 1
|
|
with open(out + ".folded", "w") as f:
|
|
f.writelines(f"{path} {us}\n" for path, us in sorted(stacks.items()))
|
|
with open(out, "w") as f:
|
|
f.write(flame_svg(stacks, n, q))
|
|
print(f"-- {len(roots)} span(s) in {len(found)} trace(s): {out}, {out}.folded", file=sys.stderr)
|
|
elif kind == "trace":
|
|
spans = trace_spans(fetch(f"http://tempo:3200/api/traces/{q}"))
|
|
kids = {}
|
|
for sp in spans:
|
|
kids.setdefault(sp.get("parentSpanId", ""), []).append(sp)
|
|
|
|
def show(sp, depth):
|
|
ms = (int(sp["endTimeUnixNano"]) - int(sp["startTimeUnixNano"])) / 1e6
|
|
attrs = {a["key"]: next(iter(a["value"].values()), None) for a in sp.get("attributes", [])
|
|
if not a["key"].startswith(("code.", "thread.")) and a["key"] not in ("busy_ns", "idle_ns", "target")}
|
|
err = " ERROR" if sp.get("status", {}).get("code") in (2, "STATUS_CODE_ERROR") else ""
|
|
print(f"{' ' * depth}{sp['name']} {ms:.0f}ms{err} {json.dumps(attrs) if attrs else ''}")
|
|
for e in sp.get("events", []):
|
|
msg = next((a["value"].get("stringValue") for a in e.get("attributes", []) if a["key"] == "message"), e.get("name"))
|
|
# A scan logs a skip for every feed not due; they bury what happened.
|
|
if '"ev":"feed_skip"' in (msg or ""):
|
|
continue
|
|
print(f"{' ' * depth} - {msg}")
|
|
for k in sorted(kids.get(sp["spanId"], []), key=lambda s: int(s["startTimeUnixNano"])):
|
|
show(k, depth + 1)
|
|
|
|
ids = {sp["spanId"] for sp in spans}
|
|
for root in [sp for sp in spans if sp.get("parentSpanId", "") not in ids]:
|
|
show(root, 0)
|
|
else:
|
|
sys.exit(__doc__)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|