SIEM Engineering & Log Pipelines

Definition

SIEM engineering covers the plumbing that makes detection possible: the log pipeline (collection agents → transport → parsing/normalization → storage → query), the normalization discipline (fields named consistently — host, user, process, src_ip — across every source, because correlation dies without it), the coverage management (which sources are logged, which are missing, which stopped — the pipeline health that every detection silently depends on), and the retention architecture (hot/warm/cold tiers, the search-performance tradeoffs, and the compliance requirements that set minimums). The engineering craft is measured at two layers: pipeline health (is the data flowing, parsed, and queryable) and query performance (can a detection or hunt run in seconds, not hours) — and both are engineering disciplines with the same rigor as any production system: monitoring, SLOs, change management.

Why It Matters

Every detection in the estate runs on the SIEM’s data, and the pipeline’s failures are silent: a parsing change breaks field normalization and every correlation quietly stops matching; an agent stops forwarding and a host goes blind with no alert; a retention tier shrinks and the 90-day hunt becomes a 7-day hunt. The detection engineering (BT9) that follows assumes normalized, complete, timely data — and the SIEM engineer is the one who delivers it. The craft’s payoff is measured in three numbers: coverage (what percentage of the estate logs, from which sources), freshness (how stale is the newest data — the detection latency floor), and query performance (whether BT7’s hunts and BT1’s correlations run interactively or overnight). The normalization discipline is the deepest value: the same field named hostname, host, computer_name, and src_host across four sources makes every correlation a four-way special case — the normalized schema (ECS-class [verify current spec] or a house standard) is what turns the SIEM from a log pile into a detection platform. For the assessor, the pipeline review is a standard engagement finding: the sources that aren’t logged (the DNS server, the IdP, the Badge PACS), the fields that don’t parse, the blind spots that explain every “we would have seen it” that didn’t happen.

Detailed Explanation

The pipeline stages and their failure modes:

 LOG PIPELINE STAGES (each can fail silently)
 [source] app/OS/appliance emits events
    |  failure: source not configured to log at all
 [agent/collector] ship to transport
    |  failure: agent dead, disk full, filter wrong
 [transport] queue/broker (kafka/syslog/ng)
    |  failure: backpressure drops, no buffering
 [parser/normalizer] raw -> schema fields
    |  failure: source update breaks parsing silently
 [storage] hot (searchable) / warm / cold (archive)
    |  failure: retention shrink, index rotation issues
 [query] detection/hunt/dashboard
    failure: schema drift breaks saved queries

 NORMALIZATION SCHEMA (the correlation substrate)
 standard fields across ALL sources:
   @timestamp, host.name, user.name, process.name,
   process.command_line, source.ip, destination.ip,
   event.action, event.outcome ...
 rule example that only works normalized:
   parent=WINWORD AND child=powershell  (BT12's chain)
   -> requires process.parent.name + process.name
      normalized across every EDR/Windows source

 PIPELINE HEALTH SLOs
 freshness:  data newest event age < 5 min (per source)
 completeness: expected sources reporting = 100%
 parse rate: unparsed/error events < 1%
 volume baseline: per-source daily volume (drift alert)

Collection starts with the source inventory: which systems exist, which log sources each emits (Windows: Security/Sysmon/PowerShell/WinRM; Linux: auth/syslog/auditd; network: flow/firewall/IDS; identity: IdP/AD; cloud: audit logs), and which are actually configured to emit — the coverage census that BT1’s “UNKNOWN owner” and every “we didn’t have that log” post-mortem feed. Parsing/normalization is the engineering core: each source gets a parser mapping raw events into the schema, and the parser suite needs its own tests (sample events per source, parsed-field assertions — the CI that catches vendor-update parsing breaks before detections do). Storage and retention are the cost/performance/compliance triangle: hot tier sized for interactive search (the hunts and correlations), warm for recent investigations, cold/archive for compliance minimums — and the tiering policy is a security decision (the BT7 hunt over 90 days requires 90 days of searchable data; the compliance requirement may demand a year in cold). Pipeline health monitoring is the discipline that makes everything else trustworthy: per-source freshness SLOs (newest-event age), completeness checks (expected sources reporting), parse-rate monitoring (unparsed-event ratio), and volume baselines (per-source daily volume with drift alerts — the agent that stops forwarding shows as a volume drop before it shows as a missed detection). Query performance is the last mile: the schema design, index strategy, and query patterns that keep the interactive workflows (BT1 triage, BT7 hunts) responsive — the SIEM engineer’s SLO to the analysts.

Common Techniques

Technique	What it does	Tooling	Notes
Source coverage census	Inventory of logged sources	Config review + agents	The blind-spot finder
Parsing/normalization suite	Raw → schema per source	SIEM parsers, custom	With test events
Schema design	Correlation substrate	ECS-class / house standard	Field naming discipline
Pipeline health monitoring	Freshness/completeness/parse-rate	Custom SLO checks	The silent-failure catch
Volume baselining	Per-source drift detection	Statistics over ingest	Agent-death early warning
Retention tiering	Cost/perf/compliance	SIEM storage config	Hunt window requirements
Query optimization	Interactive performance	Index/field strategy	The analyst’s SLO
Parser CI testing	Vendor-update resilience	Sample-event tests	Breaks caught pre-prod
Enrichment pipeline	Context at ingest	Lookups/CMDB join	Asset criticality inline
Log-source onboarding	New source process	Runbook + parser + test	Coverage growth

Full Working Code

pipehealth.py — a pipeline health checker: ingests per-source ingestion statistics (CSV: source, last_event_age_minutes, events_today, unparsed_count), compares against expectations (expected sources list, volume baselines, freshness SLOs), and reports the health matrix — stale sources, missing sources, parse-rate violations, volume drift — the SIEM engineer’s daily check, automated. stdlib only.

#!/usr/bin/env python3
"""pipehealth.py — SIEM log pipeline health checker.
Usage:
  python pipehealth.py --stats ingest_stats.csv \
      --expected expected_sources.csv --base baselines.json
ingest_stats.csv: source,last_event_age_min,events_today,unparsed
expected_sources.csv: source
baselines.json: {"source": expected_daily_volume}
stdlib only.
"""
import argparse, csv, json, sys

def load_csv(path):
    rows = []
    try:
        with open(path, encoding="utf-8", newline="") as fh:
            rows = list(csv.DictReader(fh))
    except OSError as e:
        sys.exit(f"read error: {e}")
    return rows

def check(stats, expected, baselines, fresh_max=10,
          unparsed_max_pct=1.0, drift=0.5):
    findings = []
    seen = {}
    for r in stats:
        src = r["source"].strip()
        seen[src] = r
        try:
            age = float(r["last_event_age_min"] or 0)
            events = int(r["events_today"] or 0)
            unparsed = int(r["unparsed_count"] or 0)
        except (KeyError, ValueError):
            findings.append(("HIGH", src,
                "malformed stats row"))
            continue
        if age > fresh_max:
            findings.append(("HIGH", src,
                f"stale: newest event {age:.0f} min old "
                f"(SLO {fresh_max}m)"))
        if events == 0:
            findings.append(("HIGH", src,
                "zero events today — source silent?"))
        if events and unparsed / events * 100 > \
           unparsed_max_pct:
            findings.append(("MED", src,
                f"parse failures: {unparsed}/{events} "
                f"({unparsed/events*100:.0f}%) — "
                f"parser broken by source update?"))
        base = baselines.get(src)
        if base:
            ratio = events / max(base, 1)
            if ratio < (1 - drift):
                findings.append(("MED", src,
                    f"volume drift: {events} today vs "
                    f"baseline {base} ({ratio:.0%})"))
            elif ratio > 3:
                findings.append(("LOW", src,
                    f"volume spike: {events} vs {base} — "
                    f"flooding or misconfig?"))
    for e in expected:
        src = e["source"].strip()
        if src not in seen:
            findings.append(("HIGH", src,
                "expected source NOT reporting at all"))
    findings.sort(key=lambda f: {"HIGH": 0, "MED": 1,
                                 "LOW": 2}[f[0]])
    return findings

def report(findings):
    L = ["# Pipeline health", ""]
    if not findings:
        L.append("All sources healthy.")
        return "\n".join(L) + "\n"
    L += [f"{len(findings)} findings:", ""]
    for sev, src, msg in findings:
        L.append(f"- **{sev}** `{src}` {msg}")
    L += ["", "high findings = detections silently "
          "degraded on those sources; fix first."]
    return "\n".join(L) + "\n"

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--stats", required=True)
    ap.add_argument("--expected")
    ap.add_argument("--base")
    ap.add_argument("--fresh", type=float, default=10)
    ap.add_argument("--out")
    args = ap.parse_args()
    stats = load_csv(args.stats)
    expected = load_csv(args.expected) if args.expected \
        else []
    baselines = {}
    if args.base:
        try:
            baselines = json.load(open(args.base,
                                       encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as e:
            print(f"[!] baseline skip: {e}",
                  file=sys.stderr)
    findings = check(stats, expected, baselines,
                     fresh_max=args.fresh)
    text = report(findings)
    print(text)
    if args.out:
        open(args.out, "w").write(text)
        print(f"[+] {args.out}")
    if any(f[0] == "HIGH" for f in findings):
        sys.exit(2)

if __name__ == "__main__":
    main()

Code Walkthrough

• check() (lines 26-58) runs the four health checks per source: freshness (newest-event age vs SLO — the detection-latency floor), silence (zero events — the dead source), parse rate (unparsed ratio — the vendor-update break), and volume drift (against the baseline — the agent-death early warning). The expected-sources loop (lines 59-62) catches the absent source — the one that stopped reporting so completely it has no stats row.
• The severity mapping (lines 64-66): HIGH (detections degraded — stale, silent, missing), MED (degraded quality — parse failures, drift), LOW (informational — spikes) — the triage-ready output, sorted by severity.
• The exit code (line 82): HIGH findings exit 2 — the CI/cron gate that turns the daily health check into an alerting pipeline itself (the SIEM engineer’s own monitoring, dogfooding the pattern).

How It Reaches The Target

The pipeline reaches every detection in the estate as its substrate: BT1’s triage enrichments query it, BT9’s rules match against its normalized fields, BT7’s hunts run over its retention window — and the pipehealth discipline reaches each of those consumers as the trust contract: freshness SLOs (detections fire within minutes of events), completeness guarantees (no silent blind sources), parse reliability (field assertions hold). The coverage census reaches the assessment as the blind-spot report: the sources that exist but don’t log (the IdP, the PACS, the DNS server) — every “we didn’t have that log” post-mortem finding, found before the incident. The normalization schema reaches the detection engineering as its substrate: the rule that correlates parent-child process (BT9’s workhorse) works because the schema normalizes process.parent.name across every Windows source — the engineering decision that pays on every rule after it. The volume baselines reach the operations as the early-warning layer: the agent that dies shows as a drift finding days before its blind spot becomes an incident finding. For the defender, the retention policy reaches the hunt program (BT7): the hunts that need 90 days define the hot-tier requirement — the cost conversation with data. And the parser CI reaches the vendor-update risk: every parser’s sample-event tests catch the breaking change before production does, which is the difference between a detection estate that degrades silently and one that reports its own breakage.

Exploitation Chain

1. Census — expected-sources list built: 34 sources across Windows/Linux/network/identity
2. Baseline — per-source daily volumes recorded; SLOs set (fresh 10m, parse <1%)
3. Check — pipehealth.py daily: 3 findings — IdP stale 45m, DNS zero events, FW parse 4%
4. Fix — IdP agent restart, DNS source onboarded, FW parser updated for vendor change
5. Verify — re-run: all green; freshness SLO back in tolerance
6. Detect — the DNS gap found = the logging blind spot every prior hunt missed
7. Normalize — schema pass: parent/child process fields unified across EDR+Windows
8. Enable — BT9’s parent-child rules now fire estate-wide; BT1 correlations get faster

Building Your Own Tool

schemacover.py — the schema coverage analyzer: scans parsed events (JSON-lines sample per source) against the normalization schema, reports which standard fields each source populates — the correlation-readiness matrix (which sources can join on host? on user? on process?) that tells detection engineering which rules are possible per source. stdlib only.

Design decisions

• Field-coverage matrix, not schema enforcement. The output is per-source: which standard fields are populated in what fraction of events — the matrix that answers “can I correlate this source with that one” without reading vendor docs.
• Sample-based honesty. Coverage measured on a sample (configurable N events per source) with sample size reported — a 10-event sample’s 100% coverage is a weaker claim than a 10,000-event sample’s.
• Rule-feasibility hints. The report highlights which BT9 workhorse correlations are possible per source (parent/child process, src/dst IP pairs, user/host joins) — the matrix translated into detection feasibility.
• Schema as data. The required-fields list is a JSON input — the schema evolves without touching the tool.

Full source

#!/usr/bin/env python3
"""schemacover.py — normalization schema coverage analyzer.
Usage:
  python schemacover.py --schema schema.json \
      --samples samples.jsonl
samples.jsonl: {"source": "...", "fields": {...}} per event
stdlib only.
"""
import argparse, json, sys
from collections import defaultdict

def load_schema(path):
    try:
        return json.load(open(path, encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as e:
        sys.exit(f"schema error: {e}")

def analyze(samples, schema_fields):
    per_source = defaultdict(lambda: defaultdict(int))
    counts = defaultdict(int)
    for line in samples:
        line = line.strip()
        if not line:
            continue
        try:
            ev = json.loads(line)
        except json.JSONDecodeError:
            continue
        src = ev.get("source", "?")
        counts[src] += 1
        fields = ev.get("fields", {})
        for f in schema_fields:
            if f in fields and fields[f] not in (
                    None, "", []):
                per_source[src][f] += 1
    return per_source, counts

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--schema", required=True)
    ap.add_argument("--samples", required=True)
    ap.add_argument("--out")
    args = ap.parse_args()
    schema = load_schema(args.schema)
    fields = schema.get("fields", [])
    with open(args.samples, encoding="utf-8") as fh:
        per_source, counts = analyze(fh, fields)
    if not counts:
        sys.exit("no sample events parsed")
    # correlation-feasibility matrix
    print(f"== schema coverage (schema: "
          f"{schema.get('name', '?')}) ==")
    hdr = f"{'source':<24}" + "".join(
        f"{f[:10]:>12}" for f in fields)
    print(hdr)
    print("-" * len(hdr))
    for src in sorted(counts):
        n = counts[src]
        row = f"{src:<24}"
        for f in fields:
            c = per_source[src].get(f, 0)
            row += f"{c/n:>11.0%} "
        print(row)
    print(f"\nsample sizes: "
          f"{ {s: n for s, n in counts.items()} }")
    print("\nrule feasibility (workhorse joins):")
    host_srcs = [s for s in counts
                 if per_source[s].get("host.name", 0) /
                 counts[s] > 0.9]
    user_srcs = [s for s in counts
                 if per_source[s].get("user.name", 0) /
                 counts[s] > 0.9]
    pc_srcs = [s for s in counts
               if per_source[s].get(
                   "process.parent.name", 0) /
               counts[s] > 0.9 and
               per_source[s].get("process.name", 0) /
               counts[s] > 0.9]
    print(f"  host-join capable:   {host_srcs}")
    print(f"  user-join capable:   {user_srcs}")
    print(f"  parent/child proc:   {pc_srcs}")
    if args.out:
        json.dump({"coverage": {s: dict(per_source[s])
                                for s in per_source},
                   "counts": dict(counts)},
                  open(args.out, "w"), indent=2)
        print(f"[+] {args.out}")

if __name__ == "__main__":
    main()

Walkthrough

• The coverage matrix (lines 47-56): per source, the population fraction of each schema field — the table that shows at a glance which sources are correlation-ready and which parse to half-empty records.
• The feasibility hints (lines 60-70): the three BT9 workhorse joins (host-join, user-join, parent/child process) evaluated per source at a 90% population threshold — the matrix translated into “which rules are possible,” which is the detection engineer’s scoping input.
• The sample-size honesty (line 58): sample sizes printed with the matrix — the coverage claims carry their own confidence.

Compilation and test procedure

daily pipeline health from the ingestion stats export

$ python pipehealth.py —stats ingest_stats.csv
—expected expected_sources.csv —base baselines.json

Pipeline health

3 findings:

• HIGH idp-prod stale: newest event 45 min old (SLO 10m)
• HIGH dns-edge-01 zero events today — source silent?
• MED fw-dmz parse failures: 1204/30100 (4%) — parser broken by source update? high findings = detections silently degraded on those sources; fix first.

schema coverage from parsed samples

$ python schemacover.py —schema ecs_house.json
—samples parsed_sample.jsonl == schema coverage (schema: house-ecs-v3) == source host.name user.name process… edr-agent 100% 98% 99% windows-fwd 99% 95% 97% fw-dmz 88% 0% 0% rule feasibility (workhorse joins): host-join capable: [‘edr-agent’, ‘windows-fwd’] parent/child proc: [‘edr-agent’, ‘windows-fwd’]

Detection considerations

The pipeline’s detection-adjacent role is the substrate guarantee: BT9’s rules assert on normalized fields, BT7’s hunts assume retention windows, BT1’s correlations assume joins — and the SIEM engineer’s SLOs are the contract those consumers rely on. The design controls: schema adoption (house standard or ECS-class) with the parsing suite’s CI tests, the health monitoring as a scheduled pipeline (the dogfooded check), and the onboarding runbook (new source → parser → tests → coverage review). Your own tooling hygiene: all three tools read local exports — the SIEM-side integration is the export job, not live queries at scale. Wazuh/Elastic/Splunk mappings [verify per platform]: the concepts map directly (Wazuh decoders ≈ parsers; index lifecycle ≈ retention tiers) with platform-specific implementations. The standing metrics: freshness SLO compliance per source, parse-rate per source, coverage (expected vs reporting sources), and query performance p95 for the standard hunt set — four numbers that make the pipeline as measurable as the detections it carries.

Improvements to add

1. Parser regression suite: per-source sample-event sets with field assertions, run on every parser change (the CI gate).
2. Cost-per-source reporting: ingestion volume × retention tier pricing — the cost conversation with data.
3. Auto-baseline: rolling 30-day volume medians computed automatically (the baselines.json generator).
4. Detection-impact mapping: which BT9 rules depend on which sources — the breakage blast-radius view.
5. Onboarding checklist generator: new-source runbook emitted from the source type (Windows/Linux/network templates).

Detection — Defender View

Signal	Source	ID	Notes
Source staleness	pipehealth SLO	per source	Detection-latency floor
Source silence	pipehealth	zero events	Dead agent/collection
Parse failures	pipehealth	ratio	Vendor-update breakage
Volume drift	pipehealth	vs baseline	Agent-death early warning
Missing sources	expected list diff	coverage gap	The blind-spot census
Schema coverage gaps	schemacover	per source	Correlation feasibility
Query performance p95	Hunt timings	per query class	Analyst SLO
Retention window	Storage config	policy	Hunt window requirement
Enrichment completeness	Ingest lookups	per source	Context-at-ingest
Wazuh pipeline health	Wazuh internals	[verify per deployment]	Manager/cluster checks

Evasion Techniques

The pipeline-layer evasion is adversary operations against logging — the techniques that blind the SIEM before the intrusion. Source killing: stopping agents, clearing logs (T1070.001/2 — Windows 1102 event, the alert-grade signal), disabling services — countered by pipeline health monitoring (the silence finding fires), remote log forwarding (logs leave the host before the adversary can edit), and immutable storage tiers. Volume camouflage: flooding the pipeline with noise so real events drown — countered by per-source baselines (the spike finding) and by the aggregation/dedup that separates signal classes. Parser targeting: crafting events that break parsers (malformed fields, oversized payloads) — countered by parser CI (the test events include hostile shapes) and by unparsed-rate monitoring (the parse-failure spike IS the detection). Retention blindness: adversaries pacing operations beyond the hot tier’s window — countered by the retention policy matching the hunt program (BT7’s windows define storage) and by cold-tier search capability. Field-name churn: exploiting normalization gaps (the same action logged with different field names per source) — countered by the schema discipline and the coverage matrix (the gap that shows which joins aren’t possible). The meta-point: the pipeline is the detection estate’s foundation, and its evasion surface (silence, drift, parse breaks) is exactly what the health monitoring watches — the SIEM engineering discipline is the detection estate’s immune system for its own infrastructure.

⚠️ What caught me: the first pipeline health check I built found nothing wrong — all 30 sources fresh, volumes in band — and I nearly shipped it as the “all green” baseline. Then I noticed the expected-sources file I’d written listed only the sources I knew about. The finance team’s file server had never logged to the SIEM — not broken, never onboarded — and my “all green” was a health check over a blind map. The expected-sources list isn’t a config; it’s a census (every asset that should log), and building it meant walking the CMDB against the SIEM’s source list and finding the six systems that existed outside detection entirely. Since then, the completeness check (expected vs reporting) is the check that finds the findings the freshness checks can’t — the sources that are perfectly healthy at not existing.

Lab Practice

1. Build the census: inventory your lab’s log sources (every VM, every service), write the expected-sources CSV, and verify coverage — the blind-spot census, at lab scale.
2. Baseline and drift: capture per-source daily volumes for a week, build baselines.json, then stop one agent deliberately and confirm pipehealth catches the drift before the silence.
3. Schema discipline: define a 10-field house schema, write parsers for three heterogeneous sources (syslog, Windows events, an app log), and verify with schemacover that all three correlate on host and user.
4. Parser CI: create sample-event test sets per source (including malformed/hostile shapes), wire them into a check that runs on parser changes — the vendor-update resilience, built.
5. Retention decision: compute the storage cost of 30/90/365-day hot retention for your lab’s volume, and write the tiering policy that balances BT7’s hunt windows against the budget — the cost/perf/compliance triangle, with numbers.

Interview Questions

• Technical: Walk the log pipeline stages and name the silent failure mode of each — and which health check catches it.
• Technical: Why does field normalization decide correlation feasibility, and what does a schema-coverage matrix tell detection engineering?
• Scenario: A detection stopped firing and the rule wasn’t changed. What pipeline causes do you check, in what order?
• Conceptual: Why is the expected-sources census a different (and harder) check than freshness monitoring — and what does each find?
• Conceptual: How do retention tiers, hunt windows, and compliance requirements interact — and who owns that decision?
• Deep-dive: Design the SIEM engineering function: source onboarding, schema governance, health SLOs, parser CI, and the cost management.
• Practical: Leadership cuts SIEM storage 40%. What do you tier, what do you keep hot, and how do you preserve the BT7 hunt windows?

Related Modules

BT1 (SOC Operations & Alert Triage) · BT9 (Detection Engineering & Sigma) · BT7 (Threat Hunting) · PE6 (Host Defense Evasion) · F1 (Networking Fundamentals)