AI-Enhanced Log Analysis and Anomaly Alert System — Part 3: Parsing and Normalizing Logs using Perl
In Parts 1 and 2 we laid the groundwork: we explored why AI‑driven log pipelines are now a must‑have for modern observability, and we stitched together a real‑time ingestion flow powered by Claude 4.6 Opus agents and GPT‑5.4 Pro parallel workers. Those sections gave you a streaming collector (Kafka‑compatible), a vector‑store for embeddings, and an alert‑generation micro‑service that talks to Slack.
Now it’s time to turn the raw, chaotic text that lands on our Kafka topic into something a machine‑learning model can actually understand. This third installment walks you through a full‑featured Perl parser‑normalizer that:
- Handles multiple log formats (syslog, Apache access, custom JSON blobs) in a single pass.
- Uses a hybrid approach—regular expressions for deterministic fields, and an on‑the‑fly LLM extractor (via GPT‑5.4 Pro) for the “wild‑card” parts.
- Outputs a clean, schema‑validated JSON document ready for downstream embedding.
Based on my technical understanding as a Lead Programmer Analyst, I’ll keep the Perl code idiomatic, testable, and easy to extend. Let’s dive.
Why Perl Still Rocks for Log Normalization
Perl earned its reputation as the “Swiss‑army knife” of text processing decades ago, and the language has evolved. In 2026 the community has embraced JSON::PP, Log::Log4perl, and async I/O via IO::Async. Those modules let us build a high‑throughput parser that can keep up with the 100 k‑msg/s rates we saw in the AI‑Powered Log Analysis: Find Incidents Faster in 2026 benchmark.
Architectural Overview
Component
Responsibility
**Kafka Consumer (Perl)**
Pull raw log lines from the `raw-logs` topic.
**Parser Engine**
Apply a cascade of regexes, Grok‑style patterns, and LLM‑based extraction.
**Normalizer**
Map fields to a canonical JSON schema, enforce types, and add metadata.
**Producer (Perl)**
Push the normalized JSON to the `normalized-logs` topic for the AI vectorizer.
The diagram below shows the data flow:
Kafka (raw-logs) → Perl Consumer → Parser/Normalizer → Kafka (normalized-logs) → AI Embedding Service → Anomaly Detector
Step 1 – Setting Up the Project Skeleton
First, create a directory structure that mirrors a typical CPAN‑style distribution. This keeps testing, documentation, and dependencies tidy.
mkdir -p log_parser/{lib,bin,t/etc}
cd log_parser
cat > lib/Log/Parser.pm <<'EOF'
package Log::Parser;
use strict;
use warnings;
use JSON::PP qw(encode_json decode_json);
use Log::Log4perl qw(:easy);
use IO::Async::Loop;
use Future::Utils qw( repeat );
use LWP::UserAgent; # For LLM calls (GPT‑5.4 Pro endpoint)
use URI::Escape qw(uri_escape);
use Try::Tiny;
use Exporter 'import';
our @EXPORT_OK = qw(parse_line);
EOF
We’ll flesh out parse_line in the next sections. The bin/consumer.pl script will glue everything together.
Step 2 – Defining a Canonical Schema
Before we start parsing, we need to agree on the shape of the normalized output. In 2026 most vendors (Energent.ai, LogicMonitor) recommend a flat JSON with a few mandatory keys:
-
timestamp(ISO‑8601, UTC) -
host(string) -
service(string) -
level(enum: DEBUG, INFO, WARN, ERROR, CRITICAL) -
message(string) -
attributes(object – any key‑value pairs extracted from the log) -
raw(original line for audit)
We’ll store the schema as a Perl hash for quick validation. Later you could replace it with a JSON‑Schema validator if you need stricter contracts.
my %SCHEMA = (
timestamp => qr/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$/,
host => qr/^[\w\.-]+$/,
service => qr/^\w+$/,
level => qr/^(DEBUG|INFO|WARN|ERROR|CRITICAL)$/,
message => qr/^.+$/,
attributes => sub { ref $_[0] eq 'HASH' },
raw => qr/^.+$/,
);
Step 3 – Regex‑Based Deterministic Parsers
Most traditional logs still follow predictable patterns. Let’s start with two common ones: classic syslog (RFC 5424) and Apache combined access logs. We’ll store each pattern in a hash of coderefs, making it trivial to add new parsers later.
my %PATTERNS = (
syslog => {
regex => qr/
^(?<timestamp>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z) # ISO‑8601
\s+
(?<host>[\w\.-]+) # hostname
\s+
(?<service>\w+): # service name
\s+
(?<level>DEBUG|INFO|WARN|ERROR|CRITICAL) # level
\s+
(?<message>.+) # free‑form text
$/x,
mapper => sub {
my $m = shift;
return {
timestamp => $m->{timestamp},
host => $m->{host},
service => $m->{service},
level => $m->{level},
message => $m->{message},
attributes=> {},
};
},
},
apache => {
regex => qr/
^(?<host>[\d\.]+) # IP
\s+\S+\s+\S+\s+ # ident, authuser (ignored)
\[(?<timestamp>[^\]]+)\]\s+ # [10/Oct/2026:13:55:36 -0700]
\"(?<request>[^\"]+)\"\s+ # \"GET /index.html HTTP/1.1\"
(?<status>\d{3})\s+ # 200
(?<bytes>\d+|-)\s+ # 1234
\"(?<referrer>[^\"]*)\"\s+ # \"http://example.com\"
\"(?<agent>[^\"]*)\" # \"Mozilla/5.0\"
$/x,
mapper => sub {
my $m = shift;
# Convert Apache timestamp to ISO‑8601
my $dt = DateTime::Format::Strptime->new(
pattern => '%d/%b/%Y:%H:%M:%S %z',
on_error => 'croak',
)->parse_datetime($m->{timestamp});
return {
timestamp => $dt->iso8601().'Z',
host => $m->{host},
service => 'apache',
level => $m->{status} =~ /^[45]/ ? 'ERROR' : 'INFO',
message => $m->{request},
attributes=> {
status => $m->{status}+0,
bytes => $m->{bytes} eq '-' ? 0 : $m->{bytes}+0,
referrer => $m->{referrer},
agent => $m->{agent},
},
};
},
},
);
Notice the use of named capture groups (?<name>) – a Perl feature that makes the mapper code clean and self‑documenting.
Step 4 – When Regexes Fail: LLM‑Assisted Extraction
Real‑world logs mutate. New micro‑services drop JSON blobs, legacy apps sprinkle free‑form stack traces, and container orchestrators prepend metadata that never existed before. Rather than constantly rewriting regexes, we can delegate the “guesswork” to an LLM.
Claude 4.6 Opus gave us a robust extract_fields function in Part 2. Here we’ll call the GPT‑5.4 Pro “log‑extractor” endpoint via a simple HTTP POST. The prompt is engineered to be short (to keep latency low) and to request a JSON map of timestamp, host, service, level, message, attributes. The response is then merged with any deterministic fields we already have.
my $LLM_ENDPOINT = 'https://api.openai.com/v1/chat/completions';
my $LLM_MODEL = 'gpt-5.4-pro-parallel';
my $LLM_TOKEN = $ENV{OPENAI_API_KEY};
sub llm_extract {
my ($raw_line) = @_;
my $ua = LWP::UserAgent->new( timeout => 5 );
my $payload = {
model => $LLM_MODEL,
messages => [
{ role => 'system', content => 'You are a log‑parsing assistant. Return a JSON object with fields: timestamp (ISO‑8601 UTC), host, service, level (DEBUG|INFO|WARN|ERROR|CRITICAL), message, attributes (key‑value map). If a field cannot be determined, omit it.' },
{ role => 'user', content => $raw_line },
],
temperature => 0.0,
};
my $resp = $ua->post(
$LLM_ENDPOINT,
'Content-Type' => 'application/json',
'Authorization' => "Bearer $LLM_TOKEN",
Content => encode_json($payload),
);
return {} unless $resp->is_success;
my $json = decode_json($resp->decoded_content);
my $content = $json->{choices}[0]{message}{content};
# Guard against stray markdown fences
$content =~ s/^```
json\n?//;
$content =~ s/\n?
```$//;
return eval { decode_json($content) } // {};
}
We keep the LLM call asynchronous using IO::Async::Future later, but for clarity the snippet above is synchronous. In a production pipeline you would batch calls or use the parallel‑agent pattern described in Part 2 to keep latency below 100 ms.
Step 5 – The Core parse_line Function
Now we can combine the deterministic parsers, the LLM fallback, and schema validation into a single, testable routine.
sub validate_schema {
my ($hash) = @_;
for my $key (keys %SCHEMA) {
my $rule = $SCHEMA{$key};
my $value = $hash->{$key};
if (ref $rule eq 'Regexp') {
return 0 unless defined $value && $value =~ $rule;
}
elsif (ref $rule eq 'CODE') {
return 0 unless $rule->($value);
}
else {
return 0;
}
}
return 1;
}
sub parse_line {
my ($raw_line) = @_;
my $result;
# 1️⃣ Try deterministic parsers first
for my $type (keys %PATTERNS) {
my $regex = $PATTERNS{$type}{regex};
if (my %captures = ($raw_line =~ $regex)) {
$result = $PATTERNS{$type}{mapper}->(\%captures);
last;
}
}
# 2️⃣ If nothing matched, fall back to LLM
unless ($result) {
$result = llm_extract($raw_line);
}
# 3️⃣ Ensure we always keep the raw payload for audit
$result->{raw} = $raw_line;
# 4️⃣ Validate against the canonical schema
unless (validate_schema($result)) {
WARN "Schema validation failed for line: $raw_line";
# Graceful degradation: push minimal fields
$result = {
timestamp => DateTime->now->iso8601().'Z',
host => 'unknown',
service => 'unknown',
level => 'INFO',
message => $raw_line,
attributes=> {},
raw => $raw_line,
};
}
return $result;
}
Key takeaways:
- The parser tries the fast, cheap regexes first – that covers the bulk of logs (≈ 80 % in most environments, as reported by Energent.ai).
- If no pattern matches, we invoke the LLM. The LLM is only a safety net, keeping cost predictable.
- Schema validation catches malformed dates, missing hosts, or unexpected data types before they corrupt the downstream vector store.
- We always retain the original line in the
rawfield – a best practice highlighted by LogicMonitor for forensic audits.
Step 6 – Wiring It Up: A Non‑Blocking Kafka Consumer
Below is a minimal yet production‑grade consumer using Kafka::Consumer (wrapped in IO::Async for back‑pressure handling). The script reads from raw-logs, normalizes each line, and writes to normalized-logs.
#!/usr/bin/env perl
use strict;
use warnings;
use FindBin;
use lib "$FindBin::Bin/../lib";
use Log::Parser qw(parse_line);
use Log::Log4perl qw(:easy);
use IO::Async::Loop;
use IO::Async::Timer::Periodic;
use Kafka::Consumer;
use Kafka::Producer;
use JSON::PP qw(encode_json);
use Try::Tiny;
Log::Log4perl->easy_init($INFO);
my $loop = IO::Async::Loop->new;
# Kafka config – keep these in env vars for security
my $BROKER = $ENV{KAFKA_BROKER} // 'localhost:9092';
my $GROUP = $ENV{KAFKA_GROUP} // 'log‑parser';
my $IN_TOPIC = 'raw-logs';
my $OUT_TOPIC = 'normalized-logs';
my $consumer = Kafka::Consumer->new(
host => $BROKER,
group => $GROUP,
topics => [$IN_TOPIC],
# Enable async fetches
fetch_max_bytes => 1_048_576,
);
my $producer = Kafka::Producer->new(
host => $BROKER,
required_acks => 1,
);
# Periodic timer to keep the loop alive
my $timer = IO::Async::Timer::Periodic->new(
interval => 0.1,
on_tick => sub {
while (my $msg = $consumer->poll(0)) {
my $raw = $msg->payload;
my $norm = parse_line($raw);
my $payload = encode_json($norm);
try {
$producer->produce(
$OUT_TOPIC,
0, # partition (auto)
$payload,
);
INFO "Normalized and forwarded log from $norm->{host}";
}
catch {
WARN "Failed to produce normalized log: $_";
};
}
},
);
$timer->start;
$loop->add($timer);
INFO "Log parser started – listening on $IN_TOPIC → $OUT_TOPIC";
$loop->run;
The loop runs forever, pulling messages in 0‑timeout polls (non‑blocking) and immediately feeding them through parse_line. Because the LLM call inside parse_line is synchronous, you may want to off‑load it to a dedicated worker pool (see Part 2 for the parallel‑agent pattern). In practice, a Future::Utils::repeat wrapper with a concurrency limit of 8 works well for a 100 k‑msg/s pipeline.
Step 7 – Unit Tests with Test::More
Never ship a parser without tests. Below is a tiny test suite that covers the two deterministic parsers and the LLM fallback (mocked).
!/usr/bin/env perl
use strict;
use warnings;
use Test::More tests => 4;
use FindBin;
use lib "$FindBin::Bin/../lib";
use Log::Parser qw(parse_line);
use JSON::PP qw(decode_json);
my $syslog = '2026-09-15T12:34:56Z myhost appsvc: INFO Service started successfully';
my $apache = '192.0.2.1 - - [15/Sep/2026:12:34:56 -0700] "GET /index.html HTTP/1.1" 200 1024 "http://ref.com" "Mozilla/5.0"';
my $unknown = 'CUSTOM_LOG_ENTRY: user=alice action=login result=success';
my $s = parse_line($syslog);
is
Originally published at https://artificial-inteligence.phptutorial.co.in
Top comments (0)