Approach:- Stream stdin line-by-line, parse each JSON log.- Maintain a deque of the last N valid records and an aggregation dict mapping path -> (total_duration_ms, count).- On append: update aggregation. If deque exceeds N, pop oldest and subtract its contribution.- At any point (end of input), compute average duration per path and output top 10 by average.python
#!/usr/bin/env python3
import sys
import json
import argparse
from collections import deque, defaultdict
import heapq
def top_endpoints_by_avg_duration(N, top_k=10):
window = deque() # store tuples: (path, duration_ms)
agg = defaultdict(lambda: [0.0, 0]) # path -> [sum_duration, count]
malformed = 0
line_no = 0
for line in sys.stdin:
line_no += 1
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
path = obj.get("path")
duration = obj.get("duration_ms")
# validate types
if path is None or duration is None:
raise ValueError("missing field")
duration = float(duration)
except (json.JSONDecodeError, ValueError, TypeError):
malformed += 1
continue
# add new record
window.append((path, duration))
agg[path][0] += duration
agg[path][1] += 1
# evict oldest if window too big
if len(window) > N:
old_path, old_duration = window.popleft()
agg[old_path][0] -= old_duration
agg[old_path][1] -= 1
if agg[old_path][1] == 0:
del agg[old_path] # keep memory small
# compute averages and pick top_k
averages = ((path, total / count) for path, (total, count) in agg.items())
top = heapq.nlargest(top_k, averages, key=lambda x: x[1])
# output results
for rank, (path, avg) in enumerate(top, 1):
print(f"{rank}. {path} — avg_duration_ms={avg:.2f}")
if malformed:
print(f"# skipped malformed lines: {malformed}", file=sys.stderr)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Top endpoints by avg duration over last N lines")
parser.add_argument("N", type=int, help="window size in lines")
args = parser.parse_args()
top_endpoints_by_avg_duration(args.N)
