Log parsing turns an incoming event into named, typed fields without losing the message that made the event useful. A collector can then route and transform those fields, while a log backend can filter them during an incident. The parser's job is not to infer every possible fact. It is to apply a documented contract and make failures visible.
For Fluxtail, keep the boundary clear. Parse application JSON, multiline records, and custom text in the source-side collector. Send supported fields such as message, timestamp, severity, service_name, host, and labels to the appropriate receiver. Fluxtail then applies its receiver normalization and makes the stored rows available to Live Tail, search, filters, MCP diagnostics, and AI-assisted investigation. There is no custom parser-rule interface or automatic recognition of arbitrary text assumed in this guide.
Start with a small output contract
Write the target schema before writing a regular expression. A useful HTTP JSON event can be this small:
{
"timestamp": "2026-09-15T18:18:04.120Z",
"severity": "ERROR",
"service_name": "checkout-api",
"host": "api-03",
"message": "payment provider timed out",
"trace_id": "7f3a2c",
"labels": {
"environment": "production",
"region": "ca-east"
}
}
Define a type for each field. Keep timestamps as timezone-aware values, identifiers as strings, status codes and durations as numbers, and labels as a small object of stable dimensions. A field that alternates between 503, "503", and null is valid JSON but an unstable query contract.
Keep field names consistent across services. Map level, log_level, and severity_text to one outbound severity field in the collector. Map service, app, and component only when the source-specific rule makes that meaning explicit. Do not globally rename every field named name to service_name.
The OpenTelemetry Logs Data Model is a useful vendor-neutral reference for event time, observed time, severity, body, resource, trace context, and attributes. It also makes an important distinction: an unknown severity can remain unspecified rather than being invented as INFO.
Parse JSON strictly
JSON is the best starting point when the producer controls its log format. Native parsing preserves numbers, booleans, arrays, and nested objects without a regular expression guessing at delimiters. Syntax alone is not enough; validate the record shape after decoding it. RFC 8259 defines interoperable JSON and excludes values such as NaN and infinity.
Save these fixtures as fixtures.log:
{"timestamp":"2026-09-15T14:18:04-04:00","level":"ERR","service_name":"checkout-api","message":"provider timeout","trace_id":"7f3a2c","labels":{"environment":"production"}}
2026-09-15T18:18:05Z WARN inventory-api cache warmup slow
{"timestamp":"2026-09-15 18:18:06","severity":"INFO","service_name":"checkout-api","message":"timezone missing"}
{"timestamp":"2026-09-15T18:18:07Z","timestamp":"2026-09-15T18:18:08Z","service_name":"checkout-api","message":"duplicate key"}
{"timestamp":"2026-09-15T18:18:09Z","service_name":"checkout-api","message":"invalid number","duration_ms":NaN}
not a recognized record
The following standard-library Python harness parses JSON first, then one deliberately narrow text format. Everything else becomes an unparsed fallback instead of disappearing:
import json
import re
import sys
from collections import Counter
from datetime import datetime, timezone
LABEL_KEYS = {"environment", "region"}
MAX_LABEL_VALUE = 128
TEXT = re.compile(
r"^(?P<timestamp>\d{4}-\d{2}-\d{2}T\S+) "
r"(?P<severity>TRACE|DEBUG|INFO|WARN|ERROR|FATAL) "
r"(?P<service_name>[a-z0-9][a-z0-9._-]*) "
r"(?P<message>.+)$"
)
SEVERITY = {
"TRACE": "TRACE",
"DEBUG": "DEBUG",
"INFO": "INFO",
"NOTICE": "INFO",
"WARN": "WARN",
"WARNING": "WARN",
"ERR": "ERROR",
"ERROR": "ERROR",
"CRIT": "FATAL",
"FATAL": "FATAL",
}
class ParseFailure(ValueError):
def __init__(self, code):
self.code = code
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ParseFailure("duplicate_json_key")
result[key] = value
return result
def reject_constant(_value):
raise ParseFailure("invalid_json_number")
def utc_timestamp(value):
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
raise ParseFailure("invalid_timestamp") from None
if parsed.tzinfo is None:
raise ParseFailure("invalid_timestamp")
return parsed.astimezone(timezone.utc).isoformat(
timespec="milliseconds"
).replace("+00:00", "Z")
def normalize(data):
if not isinstance(data, dict):
raise ParseFailure("invalid_record")
if not isinstance(data.get("message"), str) or not data["message"].strip():
raise ParseFailure("missing_message")
if not isinstance(data.get("service_name"), str):
raise ParseFailure("missing_service_name")
if not isinstance(data.get("timestamp"), str):
raise ParseFailure("invalid_timestamp")
source_labels = data.get("labels", {})
if not isinstance(source_labels, dict):
raise ParseFailure("invalid_labels")
labels = {}
for key in LABEL_KEYS:
if key not in source_labels:
continue
value = source_labels[key]
if not isinstance(value, str) or len(value) > MAX_LABEL_VALUE:
raise ParseFailure("invalid_labels")
labels[key] = value
source_severity = str(data.get("severity", data.get("level", ""))).upper()
severity = SEVERITY.get(source_severity)
if source_severity and severity is None:
raise ParseFailure("unknown_severity")
result = {
"timestamp": utc_timestamp(data["timestamp"]),
"message": data["message"],
"service_name": data["service_name"],
"labels": labels,
}
if severity:
result["severity"] = severity
if isinstance(data.get("trace_id"), str):
result["trace_id"] = data["trace_id"]
return result
def fallback(raw, reason):
return {
"message": raw,
"labels": {
"parse_status": "unparsed",
"parse_error": reason,
},
}
counts = Counter()
for raw_line in sys.stdin:
raw = raw_line.rstrip("\n")
if not raw:
continue
try:
try:
source = json.loads(
raw,
object_pairs_hook=unique_object,
parse_constant=reject_constant,
)
kind = "parsed_json"
except json.JSONDecodeError:
match = TEXT.fullmatch(raw)
if not match:
raise ParseFailure("no_matching_format")
source = match.groupdict()
source["labels"] = {}
kind = "parsed_text"
record = normalize(source)
counts[kind] += 1
except ParseFailure as error:
record = fallback(raw, error.code)
counts["unparsed"] += 1
counts["records_total"] += 1
print(json.dumps(record, separators=(",", ":")))
print(json.dumps(counts, sort_keys=True), file=sys.stderr)
Run it and validate every emitted line:
python3 parse_fixture.py < fixtures.log > parsed.jsonl 2> counters.json
python3 -m json.tool counters.json
python3 -c 'import json; counts = json.load(open("counters.json")); assert counts == {"parsed_json": 1, "parsed_text": 1, "records_total": 6, "unparsed": 4}'
python3 -c 'import json, sys; rows = [json.loads(line) for line in sys.stdin if line.strip()]; assert len(rows) == 6 and rows[0]["severity"] == "ERROR" and rows[2]["labels"]["parse_status"] == "unparsed" and rows[3]["labels"]["parse_error"] == "duplicate_json_key" and rows[4]["labels"]["parse_error"] == "invalid_json_number"' < parsed.jsonl
The fixtures should produce one parsed JSON record, one parsed text record, and four unparsed records. LABEL_KEYS and MAX_LABEL_VALUE are this example's explicit application policy; adjust them to a documented schema rather than passing arbitrary decoded labels through. Add assertions in the collector's test suite for the exact counters and output fields. This harness is a teaching and fixture-validation tool, not a replacement for a maintained collector.
Treat Syslog as an envelope
Use a protocol-aware parser for Syslog rather than a single regular expression over the whole message. RFC 3164 describes the legacy BSD-style envelope. Its timestamp does not include a year or timezone, so a relay or collector needs an explicit policy and may need receipt context to place the event in time.
RFC 5424 defines PRI, version, timestamp, hostname, application name, process ID, message ID, structured data, and an optional message. Those fields describe the envelope. They do not mean that the free-form MSG body is automatically parsed as JSON or key-value data.
# RFC 3164 style
<34>Sep 15 14:18:04 edge-01 sshd[1842]: Failed password
# RFC 5424 style
<34>1 2026-09-15T14:18:04-04:00 edge-01 sshd 1842 AUTH01 - Failed password
Syslog priority also needs its own mapping. It combines facility and a severity number where 0 is Emergency and 7 is Debug. That ordering is not interchangeable with every application's numeric log level. Preserve the source priority and map it with the Syslog definition instead of applying a generic integer scale.
A relay may add or replace missing timestamp and hostname information. Keep transport peer information separate from the hostname claimed inside the message when both are available. For sender configuration and TLS framing, use the dedicated Syslog forwarding guide rather than putting those transport details into a parsing rule.
Keep regex and Grok patterns narrow
Plain-text parsing works when the source format is stable and documented. Anchor the entire expression, name every captured field, restrict character classes, and leave the final human-readable message as one field. The TEXT expression above does this. It rejects a line after a format change instead of shifting tokens into the wrong columns.
Grok provides reusable names around regular expressions, but it has the same schema and backtracking risks. Expand the actual pattern during review. Test long messages, missing fields, extra spaces, Unicode, escaped delimiters, and near-matches. Put a maximum event size and execution bound at the collector or runtime layer where supported. Fluent Bit's regex parser documentation shows its named-capture and type-conversion contract; OWASP's ReDoS guidance explains why nested quantifiers and ambiguous alternation need hostile-input tests.
Do not parse values that have no investigation use. Every field becomes a compatibility commitment, and high-cardinality values used as labels can make filtering harder rather than easier. Keep request or trace identifiers in their documented fields instead of creating dynamic keys.
Join multiline events before field parsing
A stack trace split into separate input records cannot be repaired by parsing each line independently:
2026-09-15T18:18:04Z ERROR checkout-api request failed
Error: gateway timeout
at charge (/app/payment.js:42:11)
at processOrder (/app/order.js:18:5)
First join the lines into one event using source-specific start and continuation rules. Then parse the timestamp, severity, service, and combined message. Set a flush timeout and a maximum buffered size so an unterminated event cannot remain open indefinitely. Instrument timeout flushes and truncation only where the collector exposes or explicitly emits those signals; do not assume generic counters exist.
Container runtimes add another layer: one logical application event can be split into Docker or CRI fragments. Fluent Bit provides built-in docker and cri multiline handling for runtime fragments and separate multiline rules for language stack traces. Its multiline documentation distinguishes those cases. Reassemble each layer once; applying two overlapping multiline filters can duplicate or rejoin content incorrectly.
Preserve malformed input safely
An unparsed fallback should keep the original message and a bounded failure reason. It should not fabricate a timestamp, severity, service, or trace ID. Route or label it so operators can measure the gap and repair the rule.
Raw preservation also has a security boundary. An invalid record can still contain a password, bearer token, session cookie, payment data, or personal data. Prevent sensitive values at the application first, redact known fields before broad forwarding, sanitize carriage returns, newlines, and delimiters in untrusted values where they could forge log boundaries, and restrict access and retention for quarantine data. Do not copy every failed record into metrics or labels; count failure categories and keep example payloads in an access-controlled sample set. The OWASP Logging Cheat Sheet covers exclusion, sanitization, and verification controls.
Track at least:
- total records by source and parser version;
- parsed JSON, parsed Syslog, and parsed text counts;
- invalid syntax and schema mismatch counts;
- unknown severity and invalid or missing timezone counts;
- multiline timeout and truncation counts;
- unparsed fallback, collector queue, retry, and dropped-record counts.
Alert on a sustained change in the failure ratio, not one expected test fixture. A deployment that changes one field type can otherwise turn a healthy-looking input rate into unusable records.
Use the shipped collector boundary for Fluxtail
Fluxtail's published Docker setup uses Fluent Bit to read Docker JSON logs, parse the outer log field, preserve the other runtime fields, add host and container context, ensure a non-empty top-level message, and send JSON through an authenticated HTTP receiver. The relevant shipped parser stage follows Fluent Bit's documented parser filter:
[FILTER]
Name parser
Match docker.*
Key_Name log
Parser json
Reserve_Data On
Preserve_Key Off
That parser only decodes JSON in log; it does not infer arbitrary application formats. Applications should emit the documented field names, or the collector should explicitly map source fields before output. For example, a source level field can be renamed to severity with a collector rule that is tested before any fallback severity is added.
Use the public Docker collector guide for the complete input, filesystem buffering, metadata, TLS, token, and output setup. For direct HTTP producers, the HTTP JSON contract accepts one object or an array and documents HTTP 202 success. The log payload reference defines the recommended top-level fields.
After ingestion, open Live Tail and compare the received row with the fixture. Confirm message, service, severity, host, labels, and time behavior before scaling the rule to more sources. Fluxtail's search and filters documentation supports stream, time, message text, service, severity, host, labels, and documented Kubernetes fields.
Stable parsing also improves scoped MCP and AI investigation: narrow the stream and time window first, then ask for exceptions or repeated errors while keeping the raw rows available for verification. It does not make an ambiguous or discarded source record reliable after the fact. The AI log diagnostics guide covers that investigation layer separately.
Create a Fluxtail account, configure the receiver that matches the source protocol, and verify a small fixture set before forwarding a full production stream.