Build a Robust Log Parser and Prefix-Operator Search
KikoffSoftware Engineer · Technical ScreenLast updated
Implement a log parser and query engine that converts raw server log lines into structured records and supports filtering via prefix-operator tokens.
Problem Statement
Modern distributed backend architectures generate thousands of log entries across distinct microservices. Your task is to build a robust log ingestion and search component with two primary responsibilities:
- Parse: Parse raw server log strings into structured
LogRecordobjects. - Filter: Query parsed records using space-delimited prefix-operator search filters.
Log Format Specification
Each valid raw log string follows this format:
[] []
- ``: ISO-8601 string (e.g.,
2023-10-31T08:15:30Z). Contains no whitespace. - ``: One of
INFO,WARN,ERROR, orDEBUGinside square brackets. - ``: Alphanumeric service name (may include hyphens) inside square brackets.
: Remaining string text. Ifcontains a token formatted asuser_id=, extract `` into an optionaluser_idfield. If not present,user_idshould beNone.
Malformed lines that do not strictly match this structure should be dropped.
Query Operators
A search query is a space-separated string of filter tokens. A record matches the query only if it satisfies all filter tokens (logical AND):
level:: Case-insensitive exact match for log level (e.g.,level:ERROR).service:: Case-insensitive prefix match on the service name (e.g.,service:auth).user:: Exact match onuser_id.- Any token without a colon prefix is treated as a case-insensitive prefix match against any word in
message.
Interface
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class LogRecord:
timestamp: str
level: str
service: str
message: str
user_id: Optional[str] = None
class LogEngine:
def parse(self, raw_logs: List[str]) -> List[LogRecord]:
"""Parses raw log strings into structured LogRecord instances."""
pass
def search(self, records: List[LogRecord], query: str) -> List[LogRecord]:
"""Returns records matching all criteria in the query string."""
pass
Examples
Example 1
logs = [
"2023-10-31T08:15:30Z [INFO] [auth-svc] User login success user_id=u_101",
"2023-10-31T08:15:35Z [ERROR] [payment-svc] Card decline user_id=u_202",
"invalid log line with no format"
]
engine = LogEngine()
records = engine.parse(logs)
# Returns 2 records (drops the invalid line).
# Record 1 has user_id='u_101', Record 2 has user_id='u_202'.
matches = engine.search(records, "level:ERROR user:u_202")
# Returns [records[1]]
Example 2
# Using free-text word prefix
matches = engine.search(records, "service:auth log")
# Returns [records[0]] because service matches 'auth*' and message contains word starting with 'log'
Constraints
- Number of raw log lines:
0 <= N <= 10^4 - Maximum length of a log line:
1000characters - Query string length:
0 <= len(query) <= 200 - Empty query returns all parsed records.
Hints
Hint 1
Use regex or split with a maxsplit limit of 3 to reliably isolate timestamp, level, service, and message without breaking message content.
Hint 2
Extract the optional user ID by scanning message tokens for the `user_id=` prefix or using a targeted regex pattern like `r'\buser_id=(\S+)'`.
Hint 3
Split query strings into individual filter tokens and evaluate each log record against all active constraints to implement AND logic.
Follow-up questions
- How would you index these logs if the dataset grew to tens of millions of records and queries required sub-second latency?
- How would you extend the query parser to support disjunctions (`OR`), negations (`NOT`), or timestamp range queries (`from:... to:...`)?
- If different microservices emit logs in disparate formats (e.g., JSON, logfmt, custom CSV), how would you architect the ingestion pipeline?
Solution walkthrough is for members
Get the full walkthrough with reference code and complexity analysis, plus Kitty Coach at 300 messages a day. Hints, the code runner and community answers stay free.