Log Collection and Parsing

The Problem: Data Scattered Across Hundreds of Systems

Every device in an environment generates logs — firewalls, servers, endpoints, applications, cloud services, network devices. These logs are stored in different formats, on different systems, with different retention policies. A SOC cannot investigate an incident if it cannot access the relevant logs. Log collection is the process of centralizing all security-relevant logs into a single platform where they can be searched, correlated, and analyzed.

Log Sources and Their Formats

Windows Event Logs

Windows generates logs in the EVTX format (binary) or can be configured to forward via Windows Event Forwarding (WEF). Key event IDs for security:

• 4624 — successful logon
• 4625 — failed logon
• 4688 — process creation (with command line if enabled)
• 4720 — user account created
• 4728 — member added to security-enabled global group
• 7045 — service installed

# Enable PowerShell script block logging (Event ID 4104)
Set-ItemProperty -Path "HKLM:\SOFTWARE\Policies\Microsoft\Windows\PowerShell\ScriptBlockLogging" -Name "EnableScriptBlockLogging" -Value 1

# Enable process creation command line logging (Event ID 4688)
auditpol /set /subcategory:"Process Creation" /success:enable

Linux Syslog and Journald

Linux systems log to /var/log/ (syslog, auth.log, kern.log) or use systemd's journald. Syslog uses a standard format but the message content varies by application. Journald stores logs in a binary format that can be exported as JSON.

# Export journald logs as JSON
journalctl -o json > /var/log/journal-export.json

# Forward syslog to a remote collector
# In /etc/rsyslog.conf
*.* @10.0.1.10:514

Firewall and Network Device Logs

Firewalls (Palo Alto, Fortinet, Cisco ASA) log allowed and denied connections, VPN sessions, and threat detections. These are typically syslog or CEF (Common Event Format) formatted. Network devices (switches, routers, wireless controllers) log authentication, configuration changes, and interface status.

Cloud Service Logs

AWS CloudTrail, Azure Activity Logs, GCP Audit Logs, and SaaS application logs (Microsoft 365, Google Workspace) are accessed via APIs or streaming to a storage bucket. These logs are JSON-formatted and require API polling or event-driven collection.

# Collect AWS CloudTrail logs via boto3
import boto3

cloudtrail = boto3.client('cloudtrail')
response = cloudtrail.lookup_events(
    LookupAttributes=[{'AttributeKey': 'EventName', 'AttributeValue': 'ConsoleLogin'}],
    MaxResults=50
)
for event in response['Events']:
    print(event['CloudTrailEvent'])

Collection Methods

Agent-Based Collection

An agent is software installed on the endpoint that monitors log files, Windows Event Logs, or journald and forwards entries to the SIEM. Agents provide reliable delivery (with buffering and retry), local parsing, and enrichment at the source.

Common agents:

• Elastic Beats (Filebeat, Winlogbeat, Auditbeat) — lightweight, Elastic-native
• Splunk Universal Forwarder — Splunk-native, with load balancing and compression
• Wazuh — open-source, includes HIDS capabilities
• osquery — endpoint visibility, not a forwarder but can stream logs

# Filebeat configuration: monitor auth.log and forward to Logstash
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/auth.log
  fields:
    log_source: auth
    environment: production

output.logstash:
  hosts: ["10.0.1.20:5044"]

Agentless Collection

Some sources cannot have agents installed — network devices, IoT systems, legacy platforms. These use syslog forwarding, SNMP traps, or API polling. Syslog is the most common agentless protocol — the device sends UDP or TCP syslog messages to a central collector.

# Rsyslog server configuration for receiving logs
# /etc/rsyslog.conf
module(load="imudp")
input(type="imudp" port="514")
module(load="imtcp")
input(type="imtcp" port="514")

# Route logs by source
if $fromhost-ip == '10.0.2.10' then /var/log/remote/web-server.log

API-Based Collection

Cloud services and SaaS applications are collected via APIs. A collector polls the API at regular intervals (e.g., every 5 minutes) and retrieves new log entries. This introduces latency — the log may not appear in the SIEM for several minutes after the event occurs.

The Collection Pipeline

A complete log collection pipeline has four stages:

1. Collection — agent or forwarder reads the log source
2. Parsing — extract fields from the raw log (timestamp, source IP, user, action)
3. Normalization — map extracted fields to a common schema (ECS, CIM)
4. Enrichment — add context (asset owner, threat intel, geolocation)

Log Source → Agent/Forwarder → Parser → Normalizer → Enricher → SIEM Index

Each stage is a potential failure point. If the agent crashes, logs are lost. If the parser fails, fields are missing. If the normalizer maps incorrectly, queries return wrong results. Monitoring the pipeline is as important as monitoring the logs themselves.

From Raw Text to Structured Data

Raw logs are human-readable but machine-unfriendly. A syslog line like Sep 30 14:23:01 web-01 sshd[1234]: Failed password for root from 192.168.1.100 port 54321 ssh2 contains valuable information, but a SIEM cannot query "all failed SSH passwords" unless the fields are extracted and structured.

Parsing is the process of extracting fields from raw log text. Normalization is the process of mapping those fields to a standard schema so that logs from different sources can be queried uniformly.

Parsing Techniques

Regular Expressions

The most common parsing technique. A regex pattern matches the log format and captures fields into named groups.

import re

# Parse an SSH auth log
pattern = r'(?P<timestamp>\w+ \d+ \d+:\d+:\d+) (?P<host>\S+) (?P<service>\S+)\[(?P<pid>\d+)\]: (?P<message>.*)'

line = "Sep 30 14:23:01 web-01 sshd[1234]: Failed password for root from 192.168.1.100 port 54321 ssh2"
match = re.match(pattern, line)
if match:
    print(match.groupdict())
    # {'timestamp': 'Sep 30 14:23:01', 'host': 'web-01', 'service': 'sshd', 'pid': '1234', 'message': 'Failed password for root from 192.168.1.100 port 54321 ssh2'}

Grok Patterns

Grok is a pattern-matching language used in Logstash and Elastic. It composes named patterns into complex parsers.

# Grok pattern for SSH auth log
%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:host} %{WORD:service}\[%{POSINT:pid}\]: %{DATA:msg_action} %{DATA:msg_user} from %{IP:src_ip} port %{INT:src_port}

JSON Parsing

Modern applications and cloud services output JSON logs. JSON parsing is straightforward — the fields are already structured. The challenge is nested objects and arrays that need flattening for SIEM indexing.

import json

# Parse a CloudTrail JSON log
event = json.loads('{"eventTime": "2026-09-30T14:23:01Z", "eventName": "ConsoleLogin", "sourceIPAddress": "203.0.113.50", "userIdentity": {"type": "IAMUser", "userName": "admin"}}')

# Flatten nested fields
event['user_type'] = event['userIdentity']['type']
event['user_name'] = event['userIdentity']['userName']

Normalization Schemas

Elastic Common Schema (ECS)

ECS is an open-source schema that defines field names for common security event attributes. Using ECS, a Windows log and a Linux log both use source.ip for the source IP address, user.name for the username, and event.action for the action performed.

{
  "@timestamp": "2026-09-30T14:23:01.000Z",
  "event": {
    "action": "failed-authentication",
    "category": "authentication",
    "type": "start"
  },
  "source": {
    "ip": "192.168.1.100",
    "port": 54321
  },
  "user": {
    "name": "root"
  },
  "host": {
    "name": "web-01"
  }
}

Common Information Model (CIM)

Splunk's CIM is a similar normalization standard. Data models define common event types (Authentication, Network Traffic, Endpoint) with standardized field names. Searches use the CIM field names, and Splunk maps them to the actual fields in each source.

Enrichment: Adding Context to Logs

Raw logs tell you what happened. Enriched logs tell you what it matters. Enrichment adds context that makes alerts actionable:

Asset Context

Map the source IP to an asset record: hostname, owner, department, criticality, location. An alert on a critical database server is more urgent than one on a test VM.

# Enrich with asset context
def enrich_asset(ip):
    asset_db = {
        "10.0.2.10": {"hostname": "db-prod-01", "owner": "DBA team", "criticality": "high"},
        "10.0.3.15": {"hostname": "web-dev-03", "owner": "Platform team", "criticality": "low"}
    }
    return asset_db.get(ip, {"hostname": "unknown", "owner": "unknown", "criticality": "unknown"})

Threat Intelligence

Check source IPs, domains, and file hashes against threat intelligence feeds. A login from a known-malicious IP is more suspicious than one from a residential ISP.

# Enrich with threat intelligence
def check_threat_intel(ip):
    # Query AbuseIPDB, VirusTotal, or internal TI feed
    if ip in known_malicious_ips:
        return {"threat_intel": "malicious", "confidence": "high"}
    return {"threat_intel": "unknown"}

Geolocation

Map IP addresses to geographic locations. A login from a country where the company has no operations is suspicious.

import geoip2.database

reader = geoip2.database.Reader('/path/to/GeoLite2-City.mmdb')
response = reader.city('203.0.113.50')
print(response.country.name, response.city.name)

Common Parsing Failures

Timestamp Issues

Logs from different sources use different time formats and timezones. If timestamps are not normalized to UTC, correlation across sources fails. A login at 14:23:01 UTC and a file access at 09:23:01 EST are the same event, but a SIEM that does not normalize timezones treats them as 5 hours apart.

Field Name Collisions

Different sources use different field names for the same concept. Windows uses EventCode, syslog uses priority, CEF uses deviceEventClassId. Without normalization, a query for "all authentication events" must know the field name for each source.

Log Truncation

Some log sources truncate long messages. A SQL injection attempt that exceeds the syslog message limit (typically 1024 bytes) is cut off mid-payload, making detection impossible. Configure log sources to use the maximum message size and ensure the collection pipeline does not truncate.

When Logs Go Missing

Log collection is not fire-and-forget. Agents crash, disks fill, network links saturate, and parsers break when log formats change. A SOC that does not monitor its collection pipeline will discover missing logs during an incident — the worst possible time.

Common Failure Modes

Agent Crashes and Resource Exhaustion

Log collection agents consume CPU, memory, and disk I/O on the endpoint. An agent that consumes too much memory can crash the endpoint or be killed by the OS. An agent that falls behind on reading logs creates a growing backlog that eventually overflows its buffer.

# Check agent status
systemctl status filebeat

# Check agent resource usage
ps aux | grep filebeat

# Check agent logs for errors
journalctl -u filebeat --since "1 hour ago"

Buffer Overflow and Data Loss

Agents buffer logs locally when the SIEM is unreachable. If the SIEM is down for an extended period, the buffer fills and the agent either drops new logs (data loss) or blocks the log source (application impact). Configure buffer sizes and overflow policies based on the criticality of the log source.

# Filebeat output configuration with buffering
output.logstash:
  hosts: ["10.0.1.20:5044"]
  bulk_max_size: 2048
  compression_level: 3
  retry_max: 5
  retry_backoff: 1s

Clock Skew

If endpoint clocks are not synchronized (via NTP), log timestamps are inaccurate. A login that actually occurred at 14:23:01 may be logged as 14:25:33 if the endpoint clock is 2.5 minutes fast. This breaks correlation — events that occurred in sequence appear out of order.

# Check NTP synchronization
ntpq -p

# Force NTP sync
sudo ntpdate -s time.nist.gov

# Check clock offset
chronyc tracking

Parser Failures

When an application updates and changes its log format, the parser breaks. Fields that were previously extracted are now missing. The SIEM shows logs with empty fields, and detection rules that depend on those fields stop firing.

# Monitor for parsing failures
# Alert when a field that should always be present is null
# In Splunk: index=logs | stats count by field_name | where count > 0 AND isnull(field_name)

Pipeline Monitoring

The collection pipeline itself must be monitored. Key metrics:

• Logs ingested per second — by source, by agent, by parser
• Agent heartbeat — is the agent alive and reporting?
• Parsing success rate — what percentage of logs are successfully parsed?
• End-to-end latency — how long from log generation to SIEM availability?
• Buffer utilization — how full is the agent's local buffer?

# Simple pipeline health check
def check_pipeline_health():
    sources = get_all_log_sources()
    for source in sources:
        last_log = get_last_log_timestamp(source)
        if (datetime.now() - last_log).total_seconds() > 300:
            alert(f"No logs from {source} in 5 minutes")
        
        parse_rate = get_parse_success_rate(source)
        if parse_rate < 0.95:
            alert(f"Parse failure rate for {source}: {1-parse_rate:.1%}")

Log Retention and Compliance

Different log sources have different retention requirements. Compliance frameworks (PCI DSS, HIPAA, SOX) mandate minimum retention periods — often 1 year for security logs, with 90 days immediately available (hot storage) and the remainder in cold storage.

A tiered retention strategy:

• Hot storage (0–90 days) — SSD-backed, fully searchable, for active investigation
• Warm storage (90 days–1 year) — disk-backed, searchable but slower, for compliance and historical analysis
• Cold storage (1–7 years) — object storage (S3, Azure Blob), archived, for legal hold and audit

// Elastic ILM (Index Lifecycle Management) policy
{
  "policy": {
    "phases": {
      "hot": { "actions": { "rollover": { "max_age": "1d", "max_size": "50gb" } } },
      "warm": { "min_age": "90d", "actions": { "forcemerge": { "max_num_segments": 1 } } },
      "cold": { "min_age": "365d", "actions": { "freeze": {} } },
      "delete": { "min_age": "2555d", "actions": { "delete": {} } }
    }
  }
}

The retention policy must balance cost against compliance and investigative needs. Deleting logs too early risks non-compliance; retaining them too long increases storage costs and legal discovery scope.