CS PAPERS DEEP READ · PAPER 71

Pivot Tracing

Mace, Roelke & Fonseca · Brown University · SOSP 2015 (Best Paper)

What this paper did

An app you use is slow today. The engineer who has to explain why can only read the logs — the running tally the program happened to keep. The catch: that tally records whatever whoever wrote the code thought to record, and the thing you want to ask about today is probably not in it.

In 2015, three researchers at Brown proposed Pivot Tracing, which turns this around: you can put a question to a live system that cannot be stopped, watch answers stream back seconds later, and take the question away when you are done. And it can answer a kind of question nobody could answer before — such as "who, right now, is eating this machine's disk?"

First, something strange

A big system is layers stacked on layers: user jobs on top, a database in the middle, storage at the bottom. The disks at the bottom are saturated and everything is queueing. You ask: who is doing this? Nobody can tell you. Storage only knows "the database asked me to read." The database only knows "some job asked me to." The job only knows it submitted work. Every layer recognizes only the station immediately above it.

It is like a parcel depot that knows only which depot a package came from. Ask it which shopper placed the order and it genuinely does not know. The maddening part is that the answer is not missing — it is shredded across several systems, and no one holds all the pieces.

The idea

The old rule was: you must decide, while writing the code, what you will one day want to look up. Guess right and you can check afterwards. Guess wrong and you must edit the code, redeploy, and wait for the failure to happen again — which is how the expensive outages get away. Pivot Tracing changes the rule: don't record it up front; ask when you want to know.

How it pulls that off

First: the probes are temporary. The program is rewritten in place while it runs; a probe goes in, does its job, and comes back out — like hanging a counter in a shop that is still open for business, with no closing for renovation.

Second, and this is the genuinely new part: let the information ride along with the request. Give every request a small backpack. As it passes through the upper layer, that layer writes in "who I am and who I am running this errand for." The request is handed on and on, and when it finally reaches storage at the bottom, the probe there opens the backpack and sees where it came from — so "this read was N bytes" and "this was ordered by that job" are matched up on the spot.

Think of a hospital: the patient carries their own chart, so the blood lab never has to run back to reception to find out who this is. What matters is that the matching is not done by time — ten thousand things happen in the same millisecond across a datacenter, so guessing by clock is guessing wrong — but by "these two were on the same journey, one before the other."

One more thing makes it cheap: the backpack carries no ledger, just a line saying "this trip belongs to job X," and the arithmetic is done along the way, with only the result sent out at the end. The cost is real too: this only works if every component on the path faithfully passes the backpack along, and if one of them drops it, the thread of causality snaps right there.

What it changed

It changed how the question is posed: from "decide before shipping what to record" to "ask whenever you want." The word everyone now uses, observability, is measured by exactly this — can you answer a question you did not anticipate. And that small backpack now sits, under the name baggage, in the standard interfaces of mainstream tracing.

One line to remember

Give each request a backpack so the top layer can write in "who I am" and the bottom layer can open it and match — and "who is eating this disk right now" gets an answer for the first time; add probes that can be installed into and removed from a running system, and monitoring turns from bookkeeping-in-advance into asking-afterwards.

Want the causal join operator, how a query compiles into probes, and the measured overhead? → Switch to the deep read

In one sentence

Pivot Tracing lets you ask ad-hoc questions of a running distributed system: it introduces a new relational operator — the happened-before join (written ⋈→) — that pairs "a number produced here" with "an identity known only over there" along the causal path of a request, answering attribution questions that cross component and machine boundaries; and it uses dynamic instrumentation to compile the query into code and weave it into live processes, to be removed when you are done. It won Best Paper at SOSP 2015.

Glossary

Distributed system: one job done by many programs on many machines; a single user request runs through several components like a relay.
Instrumentation: measurement code buried in a program so it emits numbers or events as it runs. Logs and metric counters are instrumentation.
Dynamic instrumentation: inserting or removing that measurement code in a running process, with no source change and no restart. Solaris DTrace is the classic example.
Distributed tracing: stitching one request's calls across components into a tree, so you can see where it went and how long each hop took. Google's Dapper (2010) is the industry blueprint.
Sampling: recording only a small fraction of requests to keep the cost down — at the price that the one you need to inspect probably was not kept.
Happened-before: the causal order Lamport defined in 1978 — if a could have influenced b (a precedes b in one thread, or a sent a message that b received), then a happened before b. It needs no clock.
Relational algebra / join: the database operation that pairs rows of two tables on some condition, as in "join orders to users on user ID."
Tuple: one record, e.g. (procName=MRsort, bytesRead=4096).
RPC (remote procedure call): calling a function on another machine as if it were local; the usual way components talk.
Bytecode rewriting: Java compiles to "bytecode" run by the JVM, and that bytecode can be swapped out while the program runs — which is how dynamic instrumentation is done on the JVM.
The Hadoop stack: HDFS is the distributed file system (where bytes actually hit disk), HBase a wide-table database built on it, MapReduce a batch framework, YARN the resource scheduler. The prototype and experiments all run on this stack.

Where it sits

By Jonathan Mace, Ryan Roelke and Rodrigo Fonseca at Brown University, published at SOSP 2015, where it won Best Paper; later extended into an ACM TOCS journal article and selected as a Communications of the ACM Research Highlight (March 2020). It continues two lines at once: end-to-end tracing — Fonseca's own X-Trace (2007) and Google's Dapper (2010) — and dynamic instrumentation — DTrace and Microsoft's Fay. Downstream, Facebook's Canopy (SOSP 2017, with Mace as a co-author) took the approach to industrial scale, and the baggage abstraction it isolated now appears in the standard OpenTracing / OpenTelemetry APIs.

The problem and the motivation

Before this paper, production debugging rested on three things: logs, metrics, traces. They share one hard flaw — what gets recorded is frozen at development time. You can only look up what somebody thought to record; to ask something new you must edit code, pass review, redeploy, and wait for the failure to recur. The most expensive failures recur exactly once.

Worse is the second point: information is carved up along component boundaries. The paper's opening example is concrete: a cluster of eight machines running HBase, MapReduce jobs and direct HDFS clients all at once. Now ask the question an operator asks every day — for each HDFS DataNode, how much of the current disk read throughput belongs to which high-level client?

HDFS cannot answer. It sees only its direct clients — "all of HBase," "all of MapReduce" — and has no idea which table scan or which job's task sits behind them. HBase and MapReduce know their own job names but not how many bytes each one ultimately pressed onto which disk. The two halves of the answer live in two different systems and neither can assemble them. The paper cites real issues such as HBASE-4145 to show this is not an academic invention: engineers are asking for exactly this in bug trackers.

"Then just record everything and join it afterwards" does not work either: full-fidelity recording is too voluminous to store and too expensive to produce, and cutting the cost means sampling, which means the occurrence you care about is probably not in the sample. Under the old framing, "record it all" and "afford to record" are opposites.

Fig. 1 · The attribution gap: the cost is incurred at the bottom, but "on whose behalf" is known only at the top; each layer recognizes only the station above it.

Core ideas and method

Idea one: the happened-before join — pair by causality, not by value

This is what is genuinely new. What it is: a new relational operator, written a ⋈→ b. For each record in b, it pairs in every a-record that causally precedes it.

Why a new operator is needed: an ordinary database join pairs by value — the two tables need a shared field (the same user ID). Here there is no shared field at all: the upper layer holds a job name, the bottom layer holds a byte count, and the only thing relating them is that both happened on the same execution path of the same request, one before the other. So the pairing condition cannot be equality; it has to be Lamport's happened-before.

Why not pair on timestamps: tens of thousands of events share any given millisecond in a datacenter, and the machines' clocks do not agree anyway; matching by time is matching wrongly. Happened-before needs no clock at all — it only requires that information travel along the execution path. It is the one reliable relation a distributed system gives you for free.

So "who ate my disk" becomes a query. Roughly (simplified):

From incr In DataNodeMetrics.incrBytesRead
Join cl In First(ClientProtocols) On cl -> incr
GroupBy cl.procName
Select cl.procName, SUM(incr.delta)

Line by line: start from the DataNode event "just read N more bytes"; join it to the client-protocol record that causally precedes it (-> is happened-before, and First takes only the earliest such record, i.e. the outermost originator); group by that originator's process name; sum to get the bytes attributable to each high-level client. The number comes from the very bottom, the grouping key from the very top — which is precisely the table no single system can produce. The name Pivot comes from here: rows by machine, columns by source, and swapping the grouping key gives you another view.

Idea two: baggage — run the join on the request, not at a central node

Defining the operator is the easy half; evaluating it is the hard one. To complete this join at the bottom, the bottom must know the upper record. The naive approach ships both sides to one central machine to be matched — which is the "record everything" road again: voluminous, expensive, slow.

The paper's answer is baggage: pack the handful of fields the join needs into a small container the request itself carries, and let it travel with the request across thread handoffs, work queues and RPCs. The downstream probe pulls the upstream tuple straight out of the baggage, completes the pairing and the aggregation in place, and emits only the result.

The elegance is that causality is defined by where the request went, so the baggage's propagation path is the happened-before relation itself — no graph and no clock has to be maintained on the side.

It is also cheap. Queries are rearranged at compile time so that filtering, projection and aggregation are pushed toward the source, and the baggage carries only the fields the downstream side truly needs. Measured: with no query running, baggage is empty and serializes to 0 bytes; the heaviest query in the evaluation carried 4 tuples, about 137 bytes per request.

Fig. 2 · Evaluating the happened-before join: upstream packs identity into the baggage, which rides the request down; downstream unpacks, pairs and aggregates in place, emitting only results.

Idea three: query → advice → woven into a live process

Finally, the machinery. Three concepts:

Tracepoint: a place in the code where a probe can go — a method entry or exit, an existing log site — from which local variables are visible. A tracepoint is a possibility, and costs nothing while unused.
Advice: the short instruction sequence a query compiles into. The instruction set is deliberately tiny and restricted (no loops, no arbitrary calls), so each piece of advice has a predictable cost and cannot take the host process down. There are five kinds: Observe (turn tracepoint variables into a tuple), Unpack (take upstream tuples from the baggage), Filter, Pack (put tuples into the baggage to travel on) and Emit (send a result to the user).
Agent: a resident process on each machine that weaves advice into the matching tracepoints by bytecode rewriting, and unweaves it again.

The whole path: the user writes a query in the front end → it compiles to advice → a publish-subscribe channel broadcasts it to agents across the cluster → each agent weaves it in locally → tuples emitted by tracepoints travel back over a message bus → the front end aggregates them into a pivot table, streaming, so you watch it fill in. Retract the query and the weaving is undone; the system is as it was.

The engineering cost is stated plainly: the prototype is Java, and making baggage actually propagate required changes at thread and RPC boundaries in each system — HDFS, HBase, MapReduce and YARN each took roughly 50 to 200 lines of manual modification (common interfaces such as Thread, Runnable, Callable and Queue were handled automatically with AspectJ, while RPC protocol definitions had to be extended by hand).

Fig. 3 · From one query to a live process: compiled into restricted advice, broadcast to per-machine agents that weave it in, with only aggregates flowing back.

Key results

The motivating question does get answered. On a cluster of eight machines running HBase, MapReduce and direct HDFS clients simultaneously, Pivot Tracing breaks down DataNode disk reads and writes by the high-level client process name, presented as a pivot table of rows-by-machine and columns-by-source with sparklines — exactly the view HDFS cannot produce for itself.

The case studies found three classes of root cause. Working on that Hadoop cluster with a sequence of progressively narrower queries, the authors located a software bug, a misconfiguration, and limping hardware (a disk that had slowed down without failing). Those three are precisely what logs are worst at: nothing crashes and nothing raises an error, things are merely slow — there is no exception to catch, only attribution to do.

Overhead. With no query running the cost is negligible (baggage is empty and serializes to 0 bytes), so the machinery can sit enabled and idle. With queries running, the cost is dominated by packing, unpacking and transmitting tuples: the worst case reported is a stress test driving HDFS to its limit, where throughput drops by about 16% — an upper bound under deliberate saturation, offered to show the order of the cost rather than a typical figure.

Why it matters

First, it changed how the question is posed. The old paradigm was "decide in advance what to record," the new one is "ask afterwards." The line people now draw between observability and monitoring — can you answer a question you did not anticipate — has one of its clearest sources in this paper.

Second, it formalized cross-layer attribution. Before this, engineers did the job by stuffing a request ID into their logs and grepping the pieces back together by script. Pivot Tracing points out that this is fundamentally a join along causal order, and gives it an executable definition and implementation. Turning a chore everyone was doing by hand into an operator is the most valuable kind of systems contribution.

Third, baggage escaped the paper. Mace and Fonseca went on to isolate it as a "tracing plane"; and the notion of carrying a small bag of key-values with a request across process boundaries — under that very name, baggage — is now written into the standard OpenTracing / OpenTelemetry APIs. The baggage you propagate in a modern microservice traces its lineage here. Facebook's Canopy (SOSP 2017) carried the approach to industrial scale.

Limitations and criticism

Baggage must survive the whole path, and breaks are silent. This is the practical obstacle: if any link — a thread pool, an async queue, a third-party library, a cross-language boundary — fails to propagate the baggage, causality snaps there and the join quietly under-matches rather than erroring. What you get is a table that looks fine and is missing a piece. The effort OpenTelemetry has since poured into standardizing context propagation is an attack on exactly this.
It only sees causality along the path. What it captures is the happened-before chain the request itself traversed. Interference through shared resources — two entirely unrelated requests contending for the same disk or the same lock — lies on no such chain: you can compute who consumed what, but why they slow each other down still has to be inferred another way. By the same token it cannot answer counterfactuals ("why did this not happen").
It requires a rewritable runtime. The prototype rests on JVM bytecode rewriting; ecosystems like C++ and Go offer no equally convenient mechanism, and real datacenters are usually a patchwork of languages.
Injecting executable code into production processes makes operators and security teams wince. The paper manages the risk with a restricted, loop-free, cost-predictable instruction set, but who may issue queries, how that is audited, and how tenants are isolated are questions it does not settle.
"Ask anything" still has a boundary. The tracepoint has to exist, and what you can ask is ultimately bounded by which local variables are visible there. It enlarges the space of questions; it does not abolish its edges.
Adoption is the most honest criticism. A decade on, the industry mainstream is still Dapper-style sampled tracing plus metrics and logs; the complete form of Pivot Tracing — issuing queries at runtime and weaving them in dynamically — has not been widely deployed. What spread was its abstraction (baggage, causal propagation), not its product form.

Takeaways

1. In one line: use a happened-before join to pair "the number at the bottom" with "the identity at the top" along the causal path, and dynamic instrumentation to install the query into a running system — monitoring moves from bookkeeping-in-advance to asking-afterwards.

2. Pain one: what gets recorded is frozen at development time; a new question means editing code, redeploying, and waiting for a recurrence.

3. Pain two: information is carved along component boundaries — HDFS recognizes only direct clients and cannot say which high-level job ate the disk; the two halves live in two systems.

4. The operator: a ⋈→ b, pairing not by equal values and not by timestamps (clocks disagree) but by Lamport's causal order — the one reliable relation a distributed system gives you for free.

5. The implementation key: baggage — pack the join's fields into a container the request carries through threads, queues and RPCs; downstream unpacks and pairs in place, emitting only results instead of shipping raw events to a central node.

6. Why it is cheap: aggregation and filtering are pushed to the source and the baggage carries only what is needed — 0 bytes with no query, about 137 bytes per request for the heaviest one evaluated.

7. The machinery: query → restricted advice (Observe / Unpack / Filter / Pack / Emit) → pub-sub to agents → bytecode weaving at tracepoints → results stream back; retraction returns the cost to zero.

8. Results: DataNode disk bandwidth decomposed by high-level client on a Hadoop cluster; a software bug, a misconfiguration and a limping disk located; 50–200 lines per system to add baggage propagation; about 16% worst-case overhead under a stress test.

9. Impact: cross-layer attribution formalized as an operator; one of the sources of "observability = answering what you did not anticipate"; baggage now in the OpenTelemetry standard interfaces.

10. Limits: a broken baggage chain fails silently; only path causality is visible, not contention over shared resources; it depends on runtime rewriting and raises injection-security concerns; the complete form is still not widely adopted in industry.