TSV conversion logic for logs
Scope
- Input: arbitrary application logs in either text or JSON-lines.
- Output: tab-separated values (TSV) with a defined header. One record per log event.
- Constraints: streaming-friendly, robust to missing fields, tabs/newlines in values, mixed formats, and occasional multi-line entries.
- Target TSV schema
Use a fixed schema to maintain streaming compatibility. Recommended minimal columns:
- ts_iso: normalized timestamp in ISO 8601 (UTC if tz provided; otherwise keep naive).
- level: normalized log level (TRACE/DEBUG/INFO/WARN/ERROR/FATAL).
- service: service or application name (if discoverable).
- host: hostname or source (if discoverable).
- logger: logger name/category (if present).
- thread: thread identifier (if present).
- request_id: request or correlation id (if present).
- user_id: user identifier (if present).
- client_ip: client IP (if present).
- method: HTTP method (if present).
- path: URL path (if present).
- status: HTTP status (if present).
- latency_ms: request latency in milliseconds (if parseable).
- message: human-readable message portion.
- extras_json: JSON-encoded string of any remaining key-value pairs.
- raw: original log line for auditability.
Notes:
- If your environment requires a different set of fixed columns, adjust accordingly. Keep an extras_json catch-all for forward compatibility.
- Detection and parsing strategy
The parser operates per line and chooses one of two branches:
A) JSON log line
- Condition: line starts with “{” after leading whitespace.
- Action:
- Parse JSON into a dict; if parsing fails, treat as text log (branch B).
- Flatten to a single level: nested keys joined by dots (e.g., http.request.method → http.request.method) or use a predetermined mapping.
- Map known fields into the target schema (see section 3).
- Put unmapped fields into extras_json.
B) Text log line with optional key=value pairs
- Identify timestamp:
- Try common patterns via regex and datetime parsing:
- YYYY-MM-DD HH:MM:SS,mmm
- YYYY-MM-DD HH:MM:SS
- YYYY-MM-DDTHH:MM:SSZ
- YYYY-MM-DDTHH:MM:SS.sssZ
- Syslog-like: MMM dd HH:MM:SS (year unknown; if no year, keep as original string)
- Identify level token (case-insensitive): TRACE|DEBUG|INFO|WARN|WARNING|ERROR|CRITICAL|FATAL.
- Extract key=value pairs:
- Regex: (\w[\w.-])=("([^"\]|\.)"|\S+)
- Supports quoted values with spaces; unescape quotes and backslashes.
- Key may include dot or dash.
- Message extraction:
- Take the substring after [timestamp][optional logger/thread][level] up to the first key=value pair; trim.
- If no key=value pairs, the remainder is the message.
- Map known keys to schema; place remaining pairs in extras_json.
- Field mapping and normalization
Mapping rules (examples; adapt to your environment):
-
Timestamp (ts_iso):
- For JSON logs: prefer fields like time, timestamp, ts, @timestamp.
- For text logs: use the first recognized timestamp token.
- Normalization: convert to ISO 8601. If timezone provided, convert to UTC (e.g., 2025-09-26T12:05:34.123Z). If no timezone, output naive ISO 8601 without Z.
- If parsing fails, leave ts_iso empty and retain the original in extras_json.timestamp_raw.
-
Level:
- Map aliases: WARNING→WARN, CRITICAL→ERROR (or FATAL if preferred), VERBOSE→DEBUG.
- Output uppercase canonical values.
-
Service, host, logger, thread:
- From keys: service, app, application; host, hostname; logger, category; thread, tid. If not present, leave empty.
-
HTTP-related:
- method from method, http_method, http.request.method.
- path from path, url, uri, http.request.path.
- status from status, http.status, response.status.
- client_ip from client_ip, ip, remote_addr, http.client_ip.
-
Request/user identifiers:
- request_id from request_id, req_id, correlation_id, trace_id, x_request_id.
- user_id from user_id, uid, user.id, principal.
-
Latency:
- Accept values like “123ms”, “1.5s”, “800us”, numeric milliseconds.
- Normalize:
- ms → value
- s → value × 1000
- us/µs → value / 1000
- ns → value / 1_000_000
- If unitless integer/float, treat as milliseconds.
- If unparsable, leave latency_ms empty and persist original under extras_json.latency_raw.
-
Message:
- If JSON logs: use message, msg, log, event as primary; else let message be empty.
- For text logs: the extracted message substring.
-
extras_json:
- JSON-encode remaining unmapped key-value pairs.
- Ensure tab/newline escaping (see section 4).
-
raw:
- The original line, unmodified (except escaping tabs/newlines when outputting TSV).
- Escaping and TSV output rules
- Separator: single tab character between fields.
- Header: emit once at the start in the specified column order.
- Value sanitation:
- Replace literal tab with \t.
- Replace newline with \n and carriage return with \r.
- Remove or escape other control characters as needed.
- For extras_json, ensure it is a compact JSON string with the same escaping rules applied to embedded tabs/newlines.
- Missing values: output as empty string.
- No quoting in TSV; rely on escaping.
- Multi-line log handling
- If a line lacks a timestamp and level and does not look like JSON, treat it as a continuation of the previous record’s message.
- Append to previous message with a literal \n inserted.
- If strict one-line-per-record is required, disable multi-line merge and include the entire raw line as a separate row with empty fields.
- Example
Input lines:
- 2025-09-26 12:05:34,123 INFO request_id=abc123 user_id=42 method=GET path=/api/v1/items latency=12ms Completed request
- {"time":"2025-09-26T12:05:35.001+00:00","level":"warn","service":"checkout","http":{"method":"POST","path":"/orders"},"latency_ms":75,"message":"Payment pending"}
Output TSV (header + rows):
ts_iso level service host logger thread request_id user_id client_ip method path status latency_ms message extras_json raw
2025-09-26T12:05:34.123 INFO abc123 42 GET /api/v1/items 12 Completed request {} 2025-09-26 12:05:34,123 INFO request_id=abc123 user_id=42 method=GET path=/api/v1/items latency=12ms Completed request
2025-09-26T12:05:35.001Z WARN checkout {"http.method":"POST","http.path":"/orders"} 75 Payment pending {"http.method":"POST","http.path":"/orders"} {"time":"2025-09-26T12:05:35.001+00:00","level":"warn","service":"checkout","http":{"method":"POST","path":"/orders"},"latency_ms":75,"message":"Payment pending"}
Note: Tabs are shown conceptually above; actual output will use tab characters. Newlines/tabs inside fields must be escaped.
- Reference implementation (Python, streaming)
This implementation handles both JSON and text logs, performs normalization, and writes TSV to stdout. It avoids non-standard libraries.
import sys
import json
import re
from datetime import datetime, timezone
# Precompiled regexes
TS_PATTERNS = [
("%Y-%m-%d %H:%M:%S,%f", None),
("%Y-%m-%d %H:%M:%S", None),
("%Y-%m-%dT%H:%M:%S.%f%z", "utc"),
("%Y-%m-%dT%H:%M:%S%z", "utc"),
("%Y-%m-%dT%H:%M:%S.%fZ", "z"),
("%Y-%m-%dT%H:%M:%SZ", "z"),
]
LEVEL_RE = re.compile(r'\b(trace|debug|info|warn|warning|error|critical|fatal)\b', re.IGNORECASE)
KV_RE = re.compile(r'(\w[\w\.\-]*)=("([^"\\]|\\.)*"|\S+)')
HEADER = [
"ts_iso","level","service","host","logger","thread","request_id","user_id","client_ip",
"method","path","status","latency_ms","message","extras_json","raw"
]
def normalize_level(lvl):
if not lvl: return ""
l = lvl.upper()
if l == "WARNING": return "WARN"
if l == "CRITICAL": return "ERROR" # adjust to FATAL if preferred
return l
def parse_ts(s):
s = s.strip()
for fmt, mode in TS_PATTERNS:
try:
if "%z" in fmt:
dt = datetime.strptime(s, fmt)
if mode == "utc":
return dt.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
elif mode == "z":
# Already Zulu; ensure milliseconds where possible
if dt.microsecond:
return dt.isoformat(timespec="milliseconds").replace("+00:00", "Z")
return dt.replace(tzinfo=timezone.utc).isoformat().replace("+00:00","Z")
else:
dt = datetime.strptime(s, fmt)
# naive timestamp
if "%f" in fmt:
return dt.isoformat(timespec="milliseconds")
else:
return dt.isoformat()
except ValueError:
continue
return "" # failed
def sanitize(v):
if v is None: return ""
s = str(v)
return s.replace("\t","\\t").replace("\n","\\n").replace("\r","\\r")
def unquote(value):
# Remove surrounding quotes and unescape \" and \\ inside
if len(value) >= 2 and value[0] == '"' and value[-1] == '"':
inner = value[1:-1]
inner = inner.replace('\\"', '"').replace('\\\\', '\\')
return inner
return value
def normalize_latency(v):
if v is None: return ""
s = str(v).strip().lower()
try:
if s.endswith("ms"):
return str(float(s[:-2]))
if s.endswith("s"):
return str(float(s[:-1]) * 1000.0)
if s.endswith("us") or s.endswith("µs"):
return str(float(s[:-2]) / 1000.0)
if s.endswith("ns"):
return str(float(s[:-2]) / 1_000_000.0)
# unitless: assume milliseconds
return str(float(s))
except ValueError:
return "" # leave raw in extras_json if needed
def parse_json_line(line):
try:
obj = json.loads(line)
except json.JSONDecodeError:
return None
flat = {}
def flatten(prefix, val):
if isinstance(val, dict):
for k, v in val.items():
nk = f"{prefix}.{k}" if prefix else k
flatten(nk, v)
else:
flat[prefix] = val
flatten("", obj)
row = {k: "" for k in HEADER}
row["raw"] = line.rstrip("\n")
# Timestamp candidates
for key in ("@timestamp","time","timestamp","ts"):
if key in flat:
row["ts_iso"] = parse_ts(str(flat[key]))
if not row["ts_iso"]:
row["extras_json"] = json.dumps({"timestamp_raw": flat[key]}, separators=(",",":"))
break
row["level"] = normalize_level(flat.get("level") or flat.get("log.level") or flat.get("severity"))
row["service"] = str(flat.get("service") or flat.get("app") or flat.get("application") or "")
row["host"] = str(flat.get("host") or flat.get("hostname") or "")
row["logger"] = str(flat.get("logger") or flat.get("log.logger") or "")
row["thread"] = str(flat.get("thread") or flat.get("tid") or "")
row["request_id"] = str(flat.get("request_id") or flat.get("req_id") or flat.get("correlation_id") or flat.get("trace_id") or "")
row["user_id"] = str(flat.get("user_id") or flat.get("user.id") or flat.get("uid") or "")
row["client_ip"] = str(flat.get("client_ip") or flat.get("ip") or flat.get("remote_addr") or flat.get("http.client_ip") or "")
row["method"] = str(flat.get("method") or flat.get("http.method") or flat.get("http.request.method") or "")
row["path"] = str(flat.get("path") or flat.get("url") or flat.get("uri") or flat.get("http.path") or flat.get("http.request.path") or "")
row["status"] = str(flat.get("status") or flat.get("http.status") or flat.get("response.status") or "")
latency = flat.get("latency_ms") or flat.get("latency") or flat.get("http.latency") or None
row["latency_ms"] = normalize_latency(latency)
row["message"] = str(flat.get("message") or flat.get("msg") or flat.get("log") or flat.get("event") or "")
# Build extras from unmapped keys
mapped = set([
"@timestamp","time","timestamp","ts","level","log.level","severity","service","app","application",
"host","hostname","logger","log.logger","thread","tid","request_id","req_id","correlation_id","trace_id",
"user_id","user.id","uid","client_ip","ip","remote_addr","http.client_ip",
"method","http.method","http.request.method","path","url","uri","http.path","http.request.path",
"status","http.status","response.status","latency","latency_ms","http.latency","message","msg","log","event"
])
extras = {k: v for k, v in flat.items() if k not in mapped}
row["extras_json"] = json.dumps(extras, separators=(",",":")) if extras else "{}"
return row
def parse_text_line(line):
row = {k: "" for k in HEADER}
row["raw"] = line.rstrip("\n")
s = line.strip()
# Attempt to find timestamp at start substrings
ts_match = None
# Heuristic: check first ~40 chars for timestamp tokens separated by space or 'T'
head = s[:40]
# Try direct parsing by slicing common lengths
for token_len in (23, 19, 20, 24): # rough candidates
token = head[:token_len]
ts_iso = parse_ts(token)
if ts_iso:
row["ts_iso"] = ts_iso
ts_match = token
break
# Level
level_m = LEVEL_RE.search(s)
if level_m:
row["level"] = normalize_level(level_m.group(0))
# KV pairs
kvs = []
for m in KV_RE.finditer(s):
key = m.group(1)
val = unquote(m.group(2))
kvs.append((key, val))
# Message: substring after level up to first kv pair
msg = s
if level_m:
start = level_m.end()
msg = s[start:].strip()
if kvs:
first_kv_start = s.find(kvs[0][0] + "=")
if first_kv_start != -1:
msg = s[:first_kv_start].strip()
row["message"] = msg
# Map common keys
def get(klist):
for k in klist:
for kk, vv in kvs:
if kk == k:
return vv
return None
row["service"] = get(["service","app","application"]) or ""
row["host"] = get(["host","hostname"]) or ""
row["logger"] = get(["logger","category"]) or ""
row["thread"] = get(["thread","tid"]) or ""
row["request_id"] = get(["request_id","req_id","correlation_id","trace_id","x_request_id"]) or ""
row["user_id"] = get(["user_id","uid","user.id","principal"]) or ""
row["client_ip"] = get(["client_ip","ip","remote_addr"]) or ""
row["method"] = get(["method","http_method"]) or ""
row["path"] = get(["path","url","uri"]) or ""
row["status"] = get(["status","http_status"]) or ""
row["latency_ms"] = normalize_latency(get(["latency","latency_ms","duration","elapsed"]))
# Build extras
mapped_keys = set([
"service","app","application","host","hostname","logger","category","thread","tid",
"request_id","req_id","correlation_id","trace_id","x_request_id",
"user_id","uid","user.id","principal","client_ip","ip","remote_addr",
"method","http_method","path","url","uri","status","http_status","latency","latency_ms","duration","elapsed"
])
extras = {k: v for (k, v) in kvs if k not in mapped_keys}
row["extras_json"] = json.dumps(extras, separators=(",",":")) if extras else "{}"
return row
def emit_tsv_row(row):
fields = [sanitize(row.get(col, "")) for col in HEADER]
sys.stdout.write("\t".join(fields) + "\n")
def main():
# Write header
sys.stdout.write("\t".join(HEADER) + "\n")
for line in sys.stdin:
stripped = line.lstrip()
parsed = None
if stripped.startswith("{"):
parsed = parse_json_line(line)
if parsed is None:
parsed = parse_text_line(line)
emit_tsv_row(parsed)
if __name__ == "__main__":
main()
- Operational guidance
- Run as: python log_to_tsv.py < input.log > output.tsv
- For large files, consider:
- Using buffered I/O (default in Python is sufficient for most cases).
- Precompiling all regexes (done).
- Avoiding per-line JSON dumps for empty extras (use "{}" constant as shown).
- Parallelization: sharding input files and merging outputs; ensure identical headers.
- Monitoring and testing:
- Create unit tests with representative samples for each log format you encounter.
- Validate TSV with a reader that checks column counts and escaping.
- Track parse failure rates (e.g., empty ts_iso or level) and revisit patterns accordingly.
This logic establishes a deterministic, streaming-safe conversion of diverse logs to TSV suitable for ingestion into data warehouses or lakehouse tables. Adjust field mappings to your domain, and keep extras_json for schema evolution.
