--- title: 'Log Aggregation and Normalization' source: 'https://academia.sh/en/courses/security-operations/log-collection-and-normalization' course: 'Security Operations and Incident Response' language: en updated: '2026-09-26T03:02:56+00:00' license: 'CC BY-SA 4.0' objectives: - 'Normalizes three separately formatted fictional logs into a single schema and separately counts mapped, leftover, and empty fields' - 'Measures the gap between when a record is created and when the event happened, per source: 1, 5, and 8 minutes' - 'Fills the unmappable actor field with lookup tables and shows that as the window widens, assigned and ambiguous record counts move in opposite directions' --- # Log Aggregation and Normalization Three separately formatted fictional logs are normalized into a single schema, and the fields that do not map are counted: 15 of 24 source fields fit an eight-field schema, 9 remain unmapped; four schema fields are empty in the network-flow log, and none of the 404 records has an actor. The sources write the same event 1, 5, and 8 minutes later. Two lookup tables fill in the actor field: 115 of 139 control-plane records convert to an account, 24 do not convert; in a five-minute window, 190 of 404 flow records get an actor assigned, 212 stay empty, and at sixty minutes the assignment rises to 208 while ambiguous records jump from 2 to 39. Of 26 events that leave a trace in two sources, 21 get collected; in a one-minute window, none do. The previous lesson's handover note put into writing a link that lived in the analyst's memory. That link being present in the log from the start — that is, the machine being able to see that two alerts belong to the same actor — was assumed in the previous lesson. This lesson removes that assumption and establishes the course's unit of measure. The measure is this: which of a record's fields fit the shared schema, which do not, and **what goes in place of the field that does not fit**. The source inventory was counted in the Endpoint and System Hardening course; here the inventory is an input. What is counted is not the source's existence, but **its linkability to the others**. - **DI7.** The regional measurement network has three log sources: a cloud control-plane record, endpoint process-creation logs, and network flow data. All three are **fictional**; the addresses come from non-routable fictional blocks. - **DI8.** The target schema has eight fields: time, source, host, address, actor, action, target, result. A field a source cannot fill stays `null` and **gets counted**. - **DI9.** A source field is in one of three states: it **maps** to the schema, it stays **leftover** because the schema has no counterpart for it, or a schema field stays **empty** because no source field can fill it. - **DI10.** A record forms at the moment the source writes it, not the moment the event happens. The model gives this gap as data, per source: 1 minute at the endpoint, 5 at network flow, 8 at the control plane. - **DI11.** Thirty fictional events leave a trace in more than one source. This link is **knowledge the model holds**; it is written into no log line. - **DI12.** The actor identity takes three forms across three sources: a portal ID, a domain account, and address alone. The lookup tables live in the asset inventory, not in the log. ## Three Formats, One Schema ```js // source.mjs — three separately formatted fictional logs are produced and normalized into one schema. // IS A MODEL. Addresses come from RFC1918 fictional blocks; no real organization or host. export const SEED = 20260402; const generator = (t) => { let s = t >>> 0; return () => (s = (s * 1664525 + 1013904223) >>> 0) / 4294967296; }; const r = generator(SEED); const randInt = (n) => Math.floor(r() * n); const pick = (d) => d[randInt(d.length)]; export const HOST = Array.from({ length: 60 }, (_, i) => `uc-${101 + i}`); export const ACCOUNT = Array.from({ length: 40 }, (_, i) => `sh-${201 + i}`); export const ADDRESS = HOST.map((_, i) => `10.40.${1 + (i % 3)}.${10 + i}`); // Portal ID: 28 accounts have a match; six of them (service identities) have none at all. const PORTAL = [...ACCOUNT.slice(0, 28).map((_, i) => `ab-${4401 + i}`), ...Array.from({ length: 6 }, (_, i) => `ab-${4429 + i}`)]; const ACTION = ["object.read", "object.write", "permission.change", "key.use"]; const PATH = ["/opt/measurement/collector", "/usr/bin/archive", "/opt/measurement/verifier"]; // Target schema: eight fields. Each source fills part of it. export const SCHEMA = ["t", "source", "host", "address", "actor", "action", "target", "result"]; // [source, format, source fields, fields mapped to schema, arrival lag min, resolution] export const SOURCE = [["control-plane", "json", 8, 6, 8, "1 ms"], ["endpoint", "key=value", 8, 5, 1, "1 s"], ["network-flow", "csv", 8, 4, 5, "1 s"]]; const DAY = Date.UTC(2026, 3, 2); const iso = (ms) => new Date(DAY + ms).toISOString(); export const cp = [], ep = [], nf = []; // three raw logs, three separate formats const tagged = { cp: [], ep: [], nf: [] }; // the model's ground truth; NOT WRITTEN to the log line // Shared event: the same work item leaves a trace in more than one source. The model knows this, // the line does not. export const SHARED = []; for (let k = 0; k < 30; k++) { const i = randInt(HOST.length), h = ACCOUNT[randInt(28)]; SHARED.push({ t: 3600e3 + randInt(50000e3), host: HOST[i], address: ADDRESS[i], account: h, portal: PORTAL[ACCOUNT.indexOf(h)], ep: r() < 0.8, nf: r() < 0.9, cp: r() < 0.7 }); } const write = (o, e) => { if (o.cp) { cp.push({ ts: iso(o.t + 480e3 + randInt(9000)), actor: { id: o.portal, type: "user" }, op: pick(ACTION), target: `bucket/measurement/${randInt(900)}`, result: r() < 0.9 ? "allowed" : "denied", srcIp: o.address, requestId: `q${randInt(1e5)}` }); tagged.cp.push(e); } if (o.ep) { ep.push(`time=${Math.floor((DAY + o.t + 60e3 + randInt(9000)) / 1000)} ` + `host=${o.host} event=process-creation user=${o.account} pid=${1000 + randInt(9000)} ` + `parent=${400 + randInt(600)} path=${pick(PATH)} ` + `hash=${randInt(1e9).toString(16).padStart(8, "0")}`); tagged.ep.push(e); } if (o.nf) { nf.push([Math.floor((DAY + o.t + 300e3 + randInt(9000)) / 1000), 1 + randInt(60), o.address, `10.90.2.${10 + randInt(40)}`, pick([443, 8443, 5671]), "outbound", 400 + randInt(9e5), 5 + randInt(900)].join(",")); tagged.nf.push(e); } }; SHARED.forEach((o, i) => write(o, i)); // Background traffic: records unrelated to any shared event. for (let k = 0; k < 480; k++) { const i = randInt(HOST.length); write({ t: randInt(86400e3), host: HOST[i], address: ADDRESS[i], account: pick(ACCOUNT), portal: pick(PORTAL), ep: r() < 0.45, nf: r() < 0.75, cp: r() < 0.25 }, -1); } // Normalization: one mapper per source. A field that cannot be filled stays null. export const normal = { "control-plane": (s) => { const o = JSON.parse(s); return { t: Date.parse(o.ts), source: "control-plane", host: null, address: o.srcIp, actor: o.actor.id, action: o.op, target: o.target, result: o.result }; }, "endpoint": (s) => { const a = Object.fromEntries(s.split(" ").map((p) => p.split("="))); return { t: Number(a.time) * 1000, source: "endpoint", host: a.host, address: null, actor: a.user, action: a.event, target: a.path, result: null }; }, "network-flow": (s) => { const [ts, , src, dst, port] = s.split(","); return { t: Number(ts) * 1000, source: "network-flow", host: null, address: src, actor: null, action: null, target: `${dst}:${port}`, result: null }; }, }; export const RAW = { "control-plane": cp.map((o) => JSON.stringify(o)), "endpoint": ep, "network-flow": nf }; const TAGGED = { "control-plane": tagged.cp, "endpoint": tagged.ep, "network-flow": tagged.nf }; const all = Object.entries(RAW).flatMap(([k, l]) => l.map((s, i) => [normal[k](s), TAGGED[k][i]])).sort((a, b) => a[0].t - b[0].t); export const EVENT = all.map((x) => x[0]); export const LABEL = all.map((x) => x[1]); // shared-event index, or -1; the line NEVER SEES this if (import.meta.url === `file://${process.argv[1]}`) { const fmt = (B) => (h) => h.map((v, j) => (j ? String(v).padStart(B[j]) : String(v).padEnd(B[0]))).join(""); const yz = fmt([18, 15, 8, 7, 14, 12, 11, 13, 14]); console.log(`model (seed ${SEED}): ${EVENT.length} records, ${SCHEMA.length}-field schema`); for (const k of ["control-plane", "endpoint", "network-flow"]) console.log(RAW[k][0]); console.log(); console.log(yz(["source", "format", "records", "fields", "mapped", "leftover", "empty", "time res.", "write lag"])); for (const [name, format, fields, mapped, lag, res] of SOURCE) { const sample = normal[name](RAW[name][0]); const filled = SCHEMA.filter((a) => a !== "source" && sample[a] !== null).length; console.log(yz([name, format, RAW[name].length, fields, mapped, fields - mapped, SCHEMA.length - 1 - filled, res, `${lag} min`])); } const t = SOURCE.reduce((a, k) => a + k[2], 0), e = SOURCE.reduce((a, k) => a + k[3], 0); console.log(`\ntotal ${t} source fields: ${e} map to the schema, ` + `${t - e} remain leftover; the actor field is absent from every network-flow record`); } ``` ``` model (seed 20260402): 787 records, 8-field schema {"ts":"2026-04-02T10:26:21.519Z","actor":{"id":"ab-4401","type":"user"},"op":"permission.change","target":"bucket/measurement/888","result":"allowed","srcIp":"10.40.1.22","requestId":"q96644"} time=1775125160 host=uc-113 event=process-creation user=sh-201 pid=4052 parent=405 path=/usr/bin/archive hash=3718b893 1775125395,7,10.40.1.22,10.90.2.18,5671,outbound,120991,443 source format records fields mapped leftover empty time res. write lag control-plane json 139 8 6 2 1 1 ms 8 min endpoint key=value 244 8 5 3 2 1 s 1 min network-flow csv 404 8 4 4 4 1 s 5 min total 24 source fields: 15 map to the schema, 9 remain leftover; the actor field is absent from every network-flow record ``` Three sample lines show three separate worlds. The control-plane record writes a nested object and keeps the actor at the `actor.id` path. The endpoint record writes plain `key=value` pairs and calls the actor `user`. The network-flow record, meanwhile, is a headerless comma-separated line: **the field name is not in the record**, only its position is. The only thing that knows the first column is time is the parser. 15 of 24 source fields fit the schema, 9 remain leftover. A leftover field is not a total loss — the raw record is kept — but it **cannot be queried** through the shared schema; since the process ID or byte count finds no place in the schema, no correlation rule can use it as a criterion. The real loss is in the `empty` column. Four schema fields are empty in the network-flow record: host, actor, action, and result. The heaviest of these is actor, because every correlation operation in this course will run along the actor axis. **None of the 404 flow records say who did it.** In the endpoint record, the result field is empty: it writes that a process ran, not whether it was permitted. The last two columns separate two different problems with time. Resolution means two sources round events to the second; two records in the same second lose their order. Write lag is more decisive: the same event lands in the record one minute later at the endpoint, five at network flow, eight at the control plane. The line sees the three records **spread across seven minutes** and cannot tell whether that spread is real or comes from transmission. ## Filling the Field That Does not Map The actor field cannot be left empty, because if it stays empty the three sources never link to each other. There are two ways to fill it, and both rely on a table outside the log: a static mapping that converts the portal ID to an account, and a time-sensitive lookup that ties the address to the host and the host to the session at that moment. ```js // merge.mjs — the field that does not map is filled with lookup tables, then records are collected. import { EVENT, LABEL, SHARED, HOST, ADDRESS, ACCOUNT } from "./source.mjs"; // Lookup tables are model data: absent from the log, they come from the asset inventory. const addrToHost = new Map(ADDRESS.map((a, i) => [a, HOST[i]])); const portalToAccount = new Map(ACCOUNT.slice(0, 28).map((h, i) => [`ab-${4401 + i}`, h])); const endpointEvents = EVENT.filter((o) => o.source === "endpoint"); // The network-flow record has no actor: an attempt is made to tie the address to the host, and // the host to the account in an endpoint record within a +/- W-minute window. function assignActor(window) { const resolved = new Array(EVENT.length).fill(null); let assigned = 0, ambiguous = 0, empty = 0, portalConverted = 0; for (const [i, o] of EVENT.entries()) { if (o.source === "endpoint") { resolved[i] = o.actor; continue; } if (o.source === "control-plane") { resolved[i] = portalToAccount.get(o.actor) || null; if (resolved[i]) portalConverted++; continue; } const host = addrToHost.get(o.address); const candidates = [...new Set(endpointEvents.filter((u) => u.host === host && Math.abs(u.t - o.t) <= window * 60e3).map((u) => u.actor))]; if (candidates.length === 1) { resolved[i] = candidates[0]; assigned++; } else if (candidates.length > 1) ambiguous++; else empty++; } return { resolved, assigned, ambiguous, empty, portalConverted }; } // Records with the actor field filled are clustered along the account axis, by a // W-minute gap rule. function cluster(resolved, window) { const groups = new Map(); for (const [i, a] of resolved.entries()) if (a) (groups.get(a) || groups.set(a, []).get(a)).push(i); const clusters = []; for (const [, l] of groups) { let s = [l[0]]; for (const i of l.slice(1)) { if (EVENT[i].t - EVENT[s[s.length - 1]].t <= window * 60e3) s.push(i); else { clusters.push(s); s = [i]; } } clusters.push(s); } const multiSource = clusters.filter((s) => new Set(s.map((i) => EVENT[i].source)).size >= 2); const collected = new Set(), mixed = multiSource.filter((s) => { const labels = new Set(s.map((i) => LABEL[i]).filter((x) => x >= 0)); for (const x of labels) if (s.filter((i) => LABEL[i] === x).length >= 2) collected.add(x); return labels.size >= 2; }); const threeSource = [...collected].filter((x) => multiSource.some((s) => new Set(s.filter((i) => LABEL[i] === x) .map((i) => EVENT[i].source)).size === 3)).length; return { clusters: multiSource.length, collected: collected.size, threeSource, mixed: mixed.length }; } const fmt = (B) => (h) => h.map((v, j) => (j ? String(v).padStart(B[j]) : String(v).padEnd(B[0]))).join(""); const yz = fmt([12, 11, 11, 9, 9, 15, 13, 11]); const nf = EVENT.filter((o) => o.source === "network-flow").length; console.log(`${EVENT.length} normalized records; the actor field is empty in ${nf} network-flow ` + `records, and in ${EVENT.filter((o) => o.source === "control-plane").length} control-plane ` + `records the actor is written as a portal ID`); const twoPlus = SHARED.filter((o) => [o.ep, o.nf, o.cp].filter(Boolean).length >= 2).length; console.log(`shared events the model knows about: ${SHARED.length}, of which ${twoPlus} leave a ` + `trace in at least two sources (the line does not see this)`); console.log(); console.log(yz(["window", "assigned", "ambiguous", "empty", "clusters", "collected", "three-source", "mixed"])); for (const w of [1, 5, 15, 60]) { const a = assignActor(w), k = cluster(a.resolved, w); console.log(yz([`${w} min`, a.assigned, a.ambiguous, a.empty, k.clusters, `${k.collected}/${twoPlus}`, k.threeSource, k.mixed])); } const a5 = assignActor(5); console.log(`\ncontrol-plane records where the portal ID converts to an account: ` + `${a5.portalConverted}; unconverted ${EVENT.filter((o) => o.source === "control-plane").length - a5.portalConverted}`); console.log(`records with the actor field filled: ${a5.resolved.filter(Boolean).length}/${EVENT.length}`); ``` ``` 787 normalized records; the actor field is empty in 404 network-flow records, and in 139 control-plane records the actor is written as a portal ID shared events the model knows about: 30, of which 26 leave a trace in at least two sources (the line does not see this) window assigned ambiguous empty clusters collected three-source mixed 1 min 3 0 401 3 0/26 0 0 5 min 190 2 212 184 21/26 13 1 15 min 195 7 202 179 21/26 13 1 60 min 208 39 157 145 20/26 12 1 control-plane records where the portal ID converts to an account: 115; unconverted 24 records with the actor field filled: 549/787 ``` The static mapping is the easy one, and even it is not complete: 115 of 139 control-plane records convert to an account, **24 cannot convert**. The six portal IDs with no match are service identities and are not tied to a person in the inventory. These 24 records show the actor field as filled in the schema, but the value inside it never matches the other two sources — a filled field is more misleading than an empty one. The time-sensitive lookup, by contrast, depends directly on the window, and the table shows the window's two-sided effect. In a one-minute window, an actor can be assigned to only 3 of the 404 flow records; since the sources' write lags are 1 and 5 minutes, there is already a four-minute gap between a flow record and an endpoint record. Widening to five minutes makes assignment jump to 190. At sixty minutes, assignment gains only 18 more records, while **ambiguous records climb from 2 to 39**: as the window widens, more than one account's session becomes visible at the same host, and it becomes impossible to say whose flow it is. The last two columns show what the measure amounts to. Of the 26 fictional events that leave a trace in at least two sources, 21 are collected in a five-minute window, 13 of them with all three sources. In a one-minute window, **none** are collected — that zero comes not from missing data, but from the sources writing at different moments. All five of the uncollected events are missing **an endpoint record**; the remaining network-flow and control-plane records are three minutes apart, so the window is not the problem. The obstacle is this: the network-flow record's actor field is empty, and the lookup that fills it comes from the endpoint record. The count of mixed clusters stays at one: in this model, the cost of widening the window is not false merging, but ambiguity. In a dense environment, that column grows. ## Summary - 15 of the three sources' 24 fields map to the eight-field schema, 9 remain leftover; a leftover field is kept but cannot be queried through the shared schema, so it cannot serve as a criterion for any rule. - Four schema fields are empty in the network-flow record, and the heaviest is actor: none of the 404 records say who did it. In the endpoint record, the result field is empty. - The same event lands in the record 1 minute later at the endpoint, 5 at network flow, 8 at the control plane; the line sees the three records spread across seven minutes and cannot tell whether that spread is real or comes from transmission. - The static mapping that converts a portal ID to an account converts 115 of 139 records, and cannot convert 24; for service identities with no match, the actor field looks filled but matches nothing. - The window works both ways: 3 assignments and zero collected events at one minute, 190 assignments and 21/26 at five minutes, 208 assignments at sixty minutes but ambiguous records climbing from 2 to 39. ## Next Step In this lesson, clustering was done with a single criterion: the same actor and a time window. This criterion produced 184 clusters, and only 21 of them correspond to an event the model knows about; the rest is the same account's ordinary work over the course of a day. A security information and event management system does not cluster this way; it writes a condition sequence — which field from which source, in what order, after what count. The next lesson runs a correlation rule over the stream normalized in this lesson and produces two numbers: how many events the rule merges into a single alert, and **what it never sees at all** outside the ones it merges.