Log pipelines

Collect, normalize, enrich; guard the blind spots.

Your city's water supply runs in four steps. Reservoirs gather water from a dozen watersheds (collection). A treatment plant brings all of it up to one standard (that step is normalization). Fluoride goes in on the way out (enrichment). Mains carry it to homes and hydrants (routing). When a pump fails somewhere upstream, no alarm rings in your kitchen. The tap runs dry, and you find out during a fire. Security telemetry (the logs and events your systems emit) works the same way, and the plumbing is called a log pipeline. It collects events from laptops, cloud control planes and Kubernetes clusters, reshapes them into one common format, attaches context, then delivers them to your SIEM (security information and event management platform, the searchable store your detections run against). A rule can only match an event that actually arrived, parsed cleanly, and carried the exact field names the rule asks for. The pipeline, not the rule, sets the ceiling on what you can detect.

Four stages, and every one of them caps your detections

Collection is the job of shippers, small agents that sit next to the logs and push them onward: Vector, Fluent Bit, or Elastic Agent. They tail files, subscribe to cloud log exports, and forward events with buffering and retry so a network blip does not cost you data. Normalization renames each vendor's fields onto one shared schema so a single rule works everywhere. sourceIPAddress in AWS CloudTrail (the record of every API call made in an AWS account), IpAddress in Windows logon events, and sourceIPs[0] in Kubernetes audit logs all become source.ip. Two schemas matter in 2026: ECS (Elastic Common Schema, now converging with OpenTelemetry's semantic conventions) and OCSF (Open Cybersecurity Schema Framework, the vendor-neutral standard that AWS Security Lake and a growing list of SIEMs read natively). Pick by where the data lands: ECS if your detections, dashboards and Sigma mappings already speak Elastic, OCSF if Security Lake or a vendor that emits it natively is your store. Switching later is cheap in the pipeline and expensive everywhere else, because the remap program is fifty lines while every rule, saved search and already-indexed event keeps the old field names until somebody reindexes them. Enrichment bolts on the context an analyst would otherwise look up by hand: geo-IP (which city and country an address maps to), asset owner, identity, threat-intel tags. An alert then arrives ready to act on instead of as a bare event. Routing decides where each event lands: expensive hot SIEM storage, cheap cold object storage, or nowhere at all. Order matters. Normalize before you enrich, enrich before you route, because each stage keys off fields the stage before it produced.

Build the spine with Vector

Vector is one static Rust binary, and its config file describes a directed graph of components. Sources produce events, transforms reshape them, sinks deliver them. It handles the unglamorous parts you would otherwise build yourself: per-sink disk buffers (opt-in) so a SIEM outage does not lose events, backpressure so one slow sink cannot eat all your memory, and opt-in end-to-end acknowledgements so an event is not deleted from the source queue until a sink confirms it landed. The config below wires two sources, each through its own normalization program, then a router, then out to a hot destination and a cold one. Run vector validate before every deploy. It compiles the whole graph, catches references to components that do not exist, and health-checks every sink.

# /etc/vector/vector.yaml
sources:
  cloudtrail:
    type: aws_s3          # reads objects announced via S3 → SQS notifications
    region: us-east-1
    sqs:
      queue_url: https://sqs.us-east-1.amazonaws.com/123456789012/cloudtrail-notify
 
  k8s_audit:
    type: file
    include: ["/var/log/kubernetes/audit/*.log"]
 
transforms:
  # One CloudTrail object is a single {"Records":[...]} document, NOT one event
  # per line. Fan the array out before anything tries to read event fields.
  ct_split:
    type: remap
    inputs: [cloudtrail]
    file: /etc/vector/ct-split.vrl
    drop_on_error: true
    reroute_dropped: true
 
  # One program per source. CloudTrail and K8s audit share no field names,
  # so a single shared program would null out everything for one of them.
  normalize_cloudtrail:
    type: remap
    inputs: [ct_split]
    file: /etc/vector/cloudtrail.vrl
    drop_on_error: true      # don't let unparsed events masquerade as normalized
    reroute_dropped: true    # failed events exit via the .dropped stream
 
  normalize_k8s:
    type: remap
    inputs: [k8s_audit]
    file: /etc/vector/k8s-audit.vrl
    drop_on_error: true
    reroute_dropped: true
 
  route:
    type: route
    inputs: [normalize_cloudtrail, normalize_k8s]
    route:
      hot: '.labels.telemetry == "security"'   # detection-relevant → SIEM
 
sinks:
  siem:
    type: elasticsearch
    inputs: [route.hot]
    endpoints: ["https://siem.internal:9200"]
 
  cold_archive:
    type: aws_s3
    inputs:
      - route._unmatched
      - ct_split.dropped
      - normalize_cloudtrail.dropped
      - normalize_k8s.dropped
    region: us-east-1
    bucket: seclogs-cold
    compression: gzip
 
# Validate the whole graph before deploy:
#   $ vector validate /etc/vector/vector.yaml
#   √ Loaded ["/etc/vector/vector.yaml"]
#   √ Component configuration
#   √ Health check "siem"
#   √ Health check "cold_archive"
#   ------------------------------------
#                              Validated

Normalization is code, so test it like code

The remap transform runs a small program written in VRL (Vector Remap Language), an expression language built for this one job. Its compiler refuses to load a program that ignores an operation which can fail. Anything fallible has to be wrapped, given a fallback with ??, or marked with !, which raises the error and stops the program for that one event. Vector counts that as an error, not an abort; abort is a separate VRL keyword with a switch of its own. A malformed log line therefore cannot take your pipeline down at 3 a.m. CloudTrail needs one step before any of that. AWS does not deliver one event per line: each object in the bucket is a single gzipped document shaped {"Records":[ ... ]}, so a program that treats .message as one record parses the wrapper quite happily and then reads null out of every field it wants. The ct_split transform above fans that array out into one event per record first. Give every source its own program while you are at it, because pointing the CloudTrail mapping at Kubernetes audit events raises no error either. verb, user.username and sourceIPs[0] are simply not the names it reads, so event.action, user.name and source.ip all land null while the event stays tagged for the SIEM. The programs below map one CloudTrail record and one Kubernetes audit line onto the same ECS field names and, importantly, keep a copy of the raw event in event.original. Grab that copy before you parse. del(.message) has already thrown the bytes away by the time most people think of it, and re-encoding the parsed object instead hands you a version with key order, whitespace and number formatting normalized: readable, but it will never hash back to the source file. CloudTrail is the awkward case, because a single record only ever existed inside the Records array, so a re-encode is the best you get per event and the S3 object itself stays your integrity anchor. That copy roughly doubles storage for those sources. Pay it anyway. Normalization always flattens some vendor detail, and for security telemetry the raw event is your forensic ground truth. Because the mapping is a program, you can unit-test it against a captured sample before it ever touches production. vector vrl runs the program on its own and prints the event that comes out.

# /etc/vector/ct-split.vrl. Runs first, on the raw S3 object.
doc      = parse_json!(string!(.message))
.message = array!(doc.Records)   # '!' errors out if the wrapper shape ever changes
. = unnest!(.message)            # an array assigned to '.' emits one event each
 
# /etc/vector/cloudtrail.vrl. ct_split has left exactly ONE record in .message.
raw = object!(.message)                # '!' raises an error and stops this event
del(.message)                          # the record moves to event.original below
."@timestamp"     = raw.eventTime
.event.kind       = "event"            # ECS-allowed value ("security" is not one)
.event.action     = raw.eventName
.event.provider   = raw.eventSource
.event.original   = encode_json(raw)   # a re-encode: ct_split already parsed the
                                       # bytes, so hash the S3 object, not this
.labels.telemetry = "security"         # routing tag; ECS labels holds custom key/values
.user.name        = raw.userIdentity.userName
.source.ip        = raw.sourceIPAddress
.cloud.account.id = raw.recipientAccountId
 
# /etc/vector/k8s-audit.vrl. Same ECS fields, completely different source names.
original = string!(.message)           # the line exactly as it arrived
raw = parse_json!(original)            # audit files are one JSON object per line
del(.message)
."@timestamp"     = raw.requestReceivedTimestamp
.event.kind       = "event"
.event.action     = raw.verb
.event.provider   = "kube-apiserver"
.event.original   = original           # the bytes, not a round-trip of the parse
.labels.telemetry = "security"
.user.name        = raw.user.username
.source.ip        = raw.sourceIPs[0]
 
# Test against a REAL captured object before deploying. Cut the sample in the
# shape ct_split hands on, one record per line under .message:
#   $ gunzip -c 123456789012_CloudTrail_us-east-1_2026-07-13T0640Z_a1b2.json.gz \
#       | jq -c '{message: .Records[]}' > cloudtrail-sample.ndjson
#   $ vector vrl --program /etc/vector/cloudtrail.vrl \
#                --input cloudtrail-sample.ndjson --print-object | jq .
#   {
#     "@timestamp": "2026-07-13T06:42:17Z",
#     "cloud": { "account": { "id": "123456789012" } },
#     "event": { "action": "ConsoleLogin", "kind": "event",
#                "original": "{\"eventVersion\":\"1.08\", ...}",
#                "provider": "signin.amazonaws.com" },
#     "labels": { "telemetry": "security" },
#     "source": { "ip": "203.0.113.44" },
#     "user": { "name": "deploy-bot" }
#   }

Enrich at ingest, not at query time

You can attach context in two places: once per event as it flows through the pipeline, or on every single query in the SIEM. Ingest-time enrichment is usually cheaper, one lookup per event instead of one join per search, and it makes the alert self-contained. The analyst reads *login from Amsterdam by deploy-bot on a production account* instead of chasing three opaque IDs. The catch is staleness. Enrichment freezes context at the moment of ingest, so if an asset changes owner next month, last month's events still carry the old owner. Facts that have to be current at the moment you read them (on-call rotation, today's asset criticality) belong at query time. Facts you want pinned to the moment the event happened belong in the pipeline. Geo-IP is the clearest case, and not because it holds still: MaxMind reships GeoLite2 twice a week and city assignments genuinely move, so resolving an address six months later answers today's question about last quarter's login. Cloud account tags go the same way. Vector does this with enrichment tables: a MaxMind GeoLite2 database, an offline file that maps IP addresses to cities and countries, declared once in the config and then queried from VRL.

# vector.yaml — declare the lookup table once
enrichment_tables:
  geo:
    type: geoip
    path: /etc/vector/GeoLite2-City.mmdb
 
# remap.vrl — one lookup per event, at ingest.
# The geoip table is queried by "ip"; lookup failure coalesces to {}.
geo = get_enrichment_table_record("geo", { "ip": .source.ip }) ?? {}
.source.geo.country_iso_code = geo.country_code
.source.geo.city_name        = geo.city_name
 
# The event that reaches the SIEM now carries its own context:
#   "source": { "ip": "203.0.113.44",
#               "geo": { "city_name": "Amsterdam", "country_iso_code": "NL" } }

Route by value, because you pay by the gigabyte

Most SIEMs bill on ingest volume, dollars per gigabyte per day, and in a typical estate the majority of what you ingest is never touched by a single detection. That billing reality is why the route transform above exists. Security-relevant events go hot and searchable. Everything else lands gzip-compressed in object storage at pennies per gigabyte-month, with a written rehydration path (how you pull it back out and make it searchable again) for the day an investigation needs it. Decoupling what you collect from what you pay to index is the whole pitch of the *telemetry pipeline* product category (Vector, Cribl). Two rules keep cost-cutting from turning into self-harm. First, solve cost by tiering, never by quietly dropping whole sources. A dropped source is a permanent blind spot that no future rule can see through. Second, sampling is acceptable for high-volume, low-signal streams like VPC flow logs (connection records from your cloud virtual network), and never acceptable for authentication or control-plane events, where the one event you sampled away is the one the incident turns on.

Monitor the pipeline as a first-class detection

Attackers turn logging off on purpose. MITRE ATT&CK, the public catalogue of attacker techniques, tracks it as T1562.008, *Impair Defenses: Disable or Modify Cloud Logs*. So a source going quiet is a finding, not an ops nuisance. Watch from both ends. On the pipeline side, enable Vector's internal_metrics source, wire it to a prometheus_exporter sink, and alert on the component_errors_total rate and on buffer growth; with the Vector API enabled (api.enabled: true), vector top gives you a live per-component throughput view while you debug. On the SIEM side, run a scheduled heartbeat query, written below in KQL (Kusto Query Language, the query language Microsoft Sentinel uses), that pages someone when a table which should always report stops reporting. Write it against a hard-coded list of the tables you expect, because summarize ... by SourceTable can only return a row for a table that still has at least one event inside the query window. The source that has been dead longest is otherwise the one that quietly falls out of the results. From inside the SIEM, an upstream failure raises no error at all. It produces nothing but silence.

// Microsoft Sentinel: the pipeline's own detection rule.
// Fire when a table that should always report has gone quiet.
let LookBack = 24h;                    // set the rule's query period to match
// A table with zero events in the window produces no row at all, so start
// from the tables that must never go quiet and join what arrived onto them.
let Expected = datatable(SourceTable: string)
    ["AWSCloudTrail", "SecurityEvent", "Syslog"];
let Seen =
    union withsource=SourceTable AWSCloudTrail, SecurityEvent, Syslog
    | where TimeGenerated > ago(LookBack)
    | summarize LastEvent = max(TimeGenerated) by SourceTable;
Expected
| join kind=leftouter Seen on SourceTable
| extend SilentFor = now() - coalesce(LastEvent, ago(LookBack))
| where SilentFor > 2h
| project SourceTable, LastEvent, SilentFor
 
// Sample result. CloudTrail stopped 5½ hours ago. Syslog sent nothing in the
// whole window, so it has no LastEvent and SilentFor reads as the full
// lookback. Nothing "erred" inside the SIEM; the events never arrived:
// SourceTable     LastEvent                 SilentFor
// --------------  ------------------------  ----------
// AWSCloudTrail   2026-07-13T04:10:33.812Z  05:37:12
// Syslog                                    1.00:00:00

Everything downstream of this layer assumes the layer works. Once events land in one place, normalized and enriched, with source.ip meaning the same thing in CloudTrail, VPN logs and endpoint telemetry, you can start asking questions that span sources: *this address failed MFA (multi-factor authentication), then created an access key, inside five minutes*. Stitching events together across sources and across time is correlation, and that is where the SIEM earns its ingest bill. It is also where the next lesson, SIEM & correlation, picks up.

Treat Vector, or Fluent Bit, or whatever shipper your vendor handed you, like an application with SLOs (service level objectives, the numeric targets you promise for a service). Track events in, events out, parse failures and buffer utilization the same way you track HTTP error rates. A disk buffer that sits near full all the time is a silent outage waiting to become real data loss the next time the SIEM hiccups. Write down the rehydration path from cold storage before you need it: bucket name, compression, schema version, and who holds the rights to restore. An incident is the worst possible moment to discover that nobody ever tested the cold path.

Try this

Validate a Vector config, then dry-run the CloudTrail program against a sample cut from a real S3 object. What you want back is a normalized event that still carries event.original, not a raw passthrough.

$ vector validate /etc/vector/vector.yaml
√ Loaded ["/etc/vector/vector.yaml"]
√ Component configuration
√ Health check "siem"
√ Health check "cold_archive"
Validated
 
# Feed the program the same shape the pipeline does: one record per line.
$ gunzip -c 123456789012_CloudTrail_us-east-1_2026-07-13T0640Z_a1b2.json.gz \
    | jq -c '{message: .Records[]}' > cloudtrail-sample.ndjson
 
$ vector vrl --program /etc/vector/cloudtrail.vrl \
    --input cloudtrail-sample.ndjson --print-object | jq '{action:.event.action, ip:.source.ip, tel:.labels.telemetry}'
{
  "action": "ConsoleLogin",
  "ip": "203.0.113.44",
  "tel": "security"
}
 
# If action is null, you fed it the {"Records":[...]} wrapper instead of a single
# record, which is exactly the failure the ct_split transform exists to prevent.

Takeaway

The pipeline sets the ceiling on your detections. Collect, normalize, enrich, route, and treat silence as an alert in its own right, because attackers switch logs off deliberately.

Next step: turn on drop_on_error with a dead-letter route, add a two-hour silent-source watchdog for the tables that should never go quiet, and unit-test your remap program with vector vrl before the next schema change ships.

Related