Katkat

Build a Robust Log Parser and Prefix-Operator Search

KikoffSoftware Engineer · Technical Screen

Last 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:

  1. Parse: Parse raw server log strings into structured LogRecord objects.
  2. 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, or DEBUG inside square brackets.
  • ``: Alphanumeric service name (may include hyphens) inside square brackets.
  • : Remaining string text. If contains a token formatted as user_id=, extract `` into an optional user_id field. If not present, user_id should be None.

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 on user_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: 1000 characters
  • 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?