Log Aggregation and Normalization

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

// 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.

// 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.