#!/usr/bin/env python3 """Ask production's Loki, Tempo or Prometheus a question, from anywhere that can run docker on Tower. query.py logs '' [since] lines, newest first query.py metric '' [since] one value per series; $range is `since` query.py traces '' [since] matching traces, slowest first query.py spans '' [since] the matching spans, each with its own time query.py trace one trace's spans, as a tree query.py prom '' [since] one value per series, now; $range is `since` query.py flame '' [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'{html.escape(tip)}' f'{text}') 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'' f'{head}{"".join(rects)}') 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()