Distributed Logging Architecture: Schema, Agents, Kafka Internals & Query Planning

event_time is when the application believed the event happened, read from the wall clock of the host that produced it. observed_time is stamped by the collector when it admits the batch. Keeping both, rather than collapsing to one, is what lets you tell a late record from a record that carries a wrong timestamp. The next two blocks spell out what each timestamp can and cannot tell you.

Subtracting is only exact when both timestamps come from the same clock. Across two hosts, the difference is the true delay plus the difference between the two clocks' errors: (t_b − t_a) = true delay + (offset_b − offset_a). If the producer's clock is 4 seconds fast, t_event is 4 seconds too large, so t_durable − t_event comes out 4 seconds too small, and the record looks 4 seconds fresher than it is, possibly even negative. If it is 4 seconds slow, the record looks 4 seconds more delayed. The subtraction is only trustworthy when you can bound that offset, for example when time sync is monitored and its error is a few milliseconds, and the delay you care about is much larger. Three rules follow. Measure server-side stages (collector, broker, indexer) with server-side timestamps, because the platform controls and monitors those clocks. Measure waiting inside one process with a monotonic clock, for example an agent attaching how long each batch sat in its spool. Treat event_time to observed_time as a sanity check, not a latency measurement, because it includes the producer's clock error.

The tenant_id field is populated from the authenticated connection identity at the collector, not copied verbatim from whatever the producer's payload claims. A payload-supplied tenant_id is, at most, a hint for client-side debugging — never an authorization input. Chapter 06 covers the admission check that enforces this.

A consumer encountering a newer schema_version than it understands degrades gracefully — reading the fields it recognizes and ignoring the rest — rather than failing closed on every record from an upgraded producer.

Whether the agent runs as a sidecar in each pod or as one DaemonSet agent per node is a separate choice from how the bytes reach it. A sidecar shares a volume or the pod's network namespace with the app, which isolates one noisy workload from another but costs memory and CPU per pod. A node agent tails the runtime's log files for every pod on the node, which is cheaper and outlives any single pod, but it shares fate across all the workloads on that node. Either way, the agent receives data through one of the patterns above; it does not reach into the application's memory. Lines still sitting in the application's own in-process queue are lost if the app crashes, which is the data-loss window the next sections keep honest.

The checkpoint means 'I will never need to re-read bytes before this offset.' That is only true once the next stage has durably taken those bytes, so the checkpoint advances when a batch is acknowledged, not when it has been read or appended to an in-memory buffer. If the agent crashes after reading but before the acknowledgment, the checkpoint is still behind and the lines are read and sent again, so delivery is at-least-once and duplicates are expected. A duplicate is handled downstream by an idempotent key such as a hash of source, file identity and start offset, which is identical on every replay. Checkpointing is also batched: one durable checkpoint write per acknowledged batch, not per line, so the cost of the fsync is amortised. Two separate durability questions are in play: the checkpoint file's own durability (temp file, fsync, atomic rename, so a crash never leaves a half-written checkpoint) and the durability of the data the checkpoint vouches for (the spool or the broker). Getting the first right does not help if the second was not true when the checkpoint moved.

Identify a file by device and inode plus a fingerprint of its first bytes, because filesystems can reuse an inode number soon after a file is deleted. After a rename-based rotation, keep reading the old file descriptor to its end before switching to the new file, and if the agent was down during rotation, finish the old file by its new name where it still exists; if it has already been deleted, record an explicit gap metric instead of pretending nothing was lost. After copytruncate the file keeps its identity but its size drops below the checkpoint, which is the signal to restart at zero, accepting that any lines written between the agent's last read and the truncation are gone. A line without a trailing newline may be half written, so it stays in memory and the checkpoint never points into the middle of a line. After a restart the agent re-reads from the last complete line boundary, and the partial line is simply read again once its newline arrives.

No agent talks to one fixed collector instance — the collector tier runs as a fleet of identical, stateless replicas behind a standard load balancer, the same way any stateless HTTP service scales. Statelessness here specifically means no collector instance holds state an agent's retry depends on — a retried batch landing on a different replica than the original attempt must succeed identically, which is what makes this tier trivially horizontally scalable under the 12× burst scenario in Failure & Scale Scenarios.

A composite key of service plus instance ID or service plus a hashed shard spreads one high-volume service's traffic across many partitions, instead of the naive service-name-only key that funnels an entire service onto one partition. A consumer group's parallelism for a single partition is capped at one consumer reading it at a time, so the hot partition — not the broker count — sets the real ceiling.

Option A acks a publish once it is queued in the collector's local memory: low latency, weak durability, because a collector crash loses it. Option B acks only after the broker commit: the partition leader has the record and every replica currently in the in-sync set has it too, and that set holds at least min.insync.replicas members. This design's reference choice is option B. The acknowledgment a producer receives must mean 'a committed broker write that survives the loss of any single broker,' not merely 'a process accepted it.' The next block shows exactly which settings make that sentence true.

Kafka's documented guarantee is that a committed record is not lost as long as at least one in-sync replica (ISR member) that has it remains alive. Four settings decide how far that goes. The replication factor (RF) sets how many copies exist. With acks=all, the leader acknowledges only after every replica currently in the ISR has the record. min.insync.replicas sets the smallest ISR size at which acks=all writes are still accepted; below it, producers get a NotEnoughReplicas error instead of an ack. unclean.leader.election.enable decides whether a replica outside the ISR may become leader when no in-sync replica is left; it defaults to false, which protects acknowledged records by keeping the partition offline instead. A realistic logging setup is RF=3, acks=all, min.insync.replicas=2, unclean election disabled, with replicas spread across racks or zones. That survives any single broker failure with no data loss and no interruption to writes. It does not mean 'RF=3 tolerates losing two brokers': two failures leave one in-sync replica, below the minimum of two, so new writes are rejected. Whether previously acknowledged records survive depends on whether that last replica held them, which min.insync.replicas=2 does not guarantee for a failure of exactly the two brokers that held the up-to-date copies. Setting min.insync.replicas=3 would guarantee it but stop writes after a single failure. Also remember that an ack means the record is replicated in broker memory and the page cache, not that it has been fsynced to disk everywhere, so a simultaneous power loss across replicas is a risk that rack or zone spread reduces but does not remove.

A hot tier using option A for the 0–7 day window, where interactive, flexible incident search matters most and volume is smallest, paired with option B's cheaper label-chunk model (or an equivalent columnar approach) for the warm/cold tiers, where volume is largest and most queries are already scoped by known labels and a time range. This is a genuine trade-off, not a universal best practice — a team that never does free-text search could reasonably run option B everywhere, and a team obsessed only with interactive search and unconcerned with cost could run option A everywhere.

Sorting purely by event_time breaks pagination the instant two records share a timestamp across shards — a stable secondary key (record_id, or a monotonic per-shard sequence) is required for the merge to produce the same order on every page request.

The scan budget and concurrency cap are applied before any shard query is dispatched, not as a timeout after the fact — this is what stops a single unbounded query from starving every other concurrent search, and is the direct fix for the query-storm failure scenario.

Secret-shaped values are stripped before a record becomes durable anywhere, not after. A redaction rule that cannot confirm a field is safe fails closed — holding back the field, or the whole record, rather than guessing it's fine.