DevOps NetworkDevOpsNetwork
Join Free
Practical infra coding

Count log levels per service in JSON-lines logs

easyLog parsing and aggregation

Problem statement

A log shipper hands you structured logs as JSON lines: one JSON object per line. Count events per service and level. Print one line per service, sorted by name, with the counts for ERROR, WARN and INFO in that order.

The data is messy, as real shipped logs are:

  • Levels arrive in mixed case, and some libraries write WARNING instead of WARN. Normalise them.
  • One line was cut in half by a buffer flush.
  • One line is valid JSON but is an array, not an object.
  • One object has no service field.

Skip bad lines, print the reason for each on stderr with its line number, and print the number skipped at the end.

events.jsonl

JSON
{"ts":"2026-09-20T10:00:01Z","service":"api","level":"INFO","msg":"request served"}
{"ts":"2026-09-20T10:00:02Z","service":"api","level":"error","msg":"upstream timeout"}
{"ts":"2026-09-20T10:00:02Z","service":"worker","level":"INFO","msg":"job 41 done"}
{"ts":"2026-09-20T10:00:03Z","service":"api","level":"WARNING","msg":"slow response"}
{"ts":"2026-09-20T10:00:04Z","service":"worker","level":"ERROR","msg":"job 42 failed"}
{"ts":"2026-09-20T10:00:05Z","service":"api","level":"INFO","msg":"request
{"ts":"2026-09-20T10:00:06Z","service":"api","level":"INFO","msg":"request served"}
["not", "an", "object"]
{"ts":"2026-09-20T10:00:07Z","level":"INFO","msg":"no service field"}
{"ts":"2026-09-20T10:00:08Z","service":"worker","level":"INFO","msg":"job 43 done"}

Examples

Example 1

Input: python solution.py events.jsonl

Output: api ERROR=1 WARN=1 INFO=2 worker ERROR=1 WARN=0 INFO=2 skipped 3 line(s)

Explanation: stderr shows line 6: invalid JSON, line 8: not a JSON object and line 9: missing service or level. error counts as ERROR and WARNING counts as WARN.

Hints

Approach

Optimal

Validate in layers and report which layer failed.

  1. Skip blank lines. Parse each line on its own; on JSONDecodeError, report e.msg with the line number.
  2. Require a dict, then require service and level to be strings. event.get returns None for missing keys, so a missing key and a null value are handled the same way.
  3. Normalise the level with .strip().upper() and map aliases such as WARNING to WARN.
  4. Increment counts[service][level]. A Counter returns 0 for levels a service never logged, so WARN=0 prints without special-casing.

Time is linear. Memory is one counter per (service, level) pair, independent of file size.

ComplexityTime O(n)Space O(s * l)
Python
import json
import sys
from collections import Counter, defaultdict
LEVELS = ["ERROR", "WARN", "INFO"]
ALIASES = {"WARNING": "WARN", "ERR": "ERROR", "FATAL": "ERROR"}
def count_levels(lines):
counts = defaultdict(Counter)
skipped = 0
for lineno, line in enumerate(lines, 1):
if not line.strip():
continue
try:
event = json.loads(line)
except json.JSONDecodeError as e:
print(f"line {lineno}: invalid JSON ({e.msg})", file=sys.stderr)
skipped += 1
continue
if not isinstance(event, dict):
print(f"line {lineno}: not a JSON object", file=sys.stderr)
skipped += 1
continue
service, level = event.get("service"), event.get("level")
if not isinstance(service, str) or not isinstance(level, str):
print(f"line {lineno}: missing service or level", file=sys.stderr)
skipped += 1
continue
level = level.strip().upper()
counts[service][ALIASES.get(level, level)] += 1
return counts, skipped
def main(path):
with open(path, encoding="utf-8", errors="replace") as f:
counts, skipped = count_levels(f)
for service in sorted(counts):
c = counts[service]
print(service, " ".join(f"{lvl}={c[lvl]}" for lvl in LEVELS))
print(f"skipped {skipped} line(s)")
if __name__ == "__main__":
main(sys.argv[1] if len(sys.argv) > 1 else "events.jsonl")

Follow-up questions

  • Also report the top 3 error messages per service.
  • Some events nest the level under log.level. How do you look up a dotted path safely?
  • The file is gzipped and 10 GB. What changes? (Only the open call.)

Frequently asked questions

stdout is the data, which another program might parse or a test might compare. stderr is for humans. Keeping them apart means python solution.py events.jsonl > report.txt gives a clean report while you still see the warnings in the terminal.

Most modern services log JSON, and anything that ships logs (Fluent Bit, Vector, a Lambda subscription) sometimes delivers truncated or odd records. The interviewer wants to see that one bad record is skipped and reported, not allowed to crash the whole batch.