Correlating Crawl Logs With Trace IDs #

A single logical fetch produces a scatter of events — the dispatch, the rules decision, the delay applied, two retries, the parse, the write — and without a shared identifier they are unrelatable. Adding one field turns three otherwise impossible questions into ordinary queries. This guide covers generating, propagating and querying trace identifiers across a crawl, as part of structured logging and log shipping in the Pipeline Storage, Deduplication & Monitoring section.

Problem Framing #

Consider a record with a suspicious value. To explain it you need to know: which URL produced it, how many attempts that URL took, what the crawler did between the first failure and the eventual success, which rules snapshot was in force, and which extractor build parsed the response. Every one of those facts exists in a log somewhere, and without a correlation key they cannot be joined.

Timestamp-and-host correlation is the usual substitute and it fails at concurrency. Twenty workers fetching the same host within the same second produce interleaved events that cannot be attributed, and retries make it worse — the same URL appears several times with no way to tell which delay belonged to which attempt.

The fix is a trace identifier generated when the URL leaves the frontier and carried through every event it causes, including into the stored record. It costs one field.

Step-by-Step Implementation #

1. Generate once, at the frontier #

import uuid

def new_trace_id() -> str:
    """One identifier per logical fetch — not per attempt, not per worker."""
    return uuid.uuid4().hex

def dispatch(frontier, url: str, host: str) -> dict:
    return {"trace_id": new_trace_id(), "url": url, "host": host, "attempt": 0}

The distinction between “per fetch” and “per attempt” is the one that matters. A trace covers the whole logical operation including its retries; an attempt counter distinguishes the individual tries within it. Generating a fresh identifier per attempt loses exactly the relationship the trace exists to express.

2. Bind it to a context, not to a parameter #

Threading trace_id through every function signature is tedious and, more importantly, fragile — it survives until someone adds a helper and forgets. A context variable binds it once and every log call picks it up.

import contextvars
from contextlib import contextmanager
import structlog

_trace = contextvars.ContextVar("trace_id", default=None)
_host = contextvars.ContextVar("host", default=None)

@contextmanager
def crawl_context(trace_id: str, host: str):
    trace_token, host_token = _trace.set(trace_id), _host.set(host)
    try:
        yield
    finally:
        _trace.reset(trace_token)
        _host.reset(host_token)

def add_correlation(_logger, _method, event: dict) -> dict:
    trace_id, host = _trace.get(), _host.get()
    if trace_id:
        event["trace_id"] = trace_id
    if host:
        event["host"] = host
    return event

structlog.configure(processors=[
    add_correlation,
    redact_event,                       # redaction before anything is emitted
    structlog.processors.TimeStamper(fmt="iso", utc=True),
    structlog.processors.JSONRenderer(),
])

Context variables work correctly under asyncio — each task gets its own copy — which is what makes this usable in a concurrent crawler where a thread-local would leak between tasks.

Ordering the processors so redaction runs before rendering is deliberate; the mechanics are covered in the structured logging topic.

3. Emit events at the decision points #

log = structlog.get_logger()

async def fetch_with_trace(task: dict, gates, transport) -> dict | None:
    with crawl_context(task["trace_id"], task["host"]):
        log.info("dispatch", url=task["url"], attempt=task["attempt"])

        decision = gates.rules.decide(task["url"])
        log.info("rules_decision", allowed=decision.allowed,
                 rules_sha256=decision.rules_sha256[:16], source=decision.source)
        if not decision.allowed:
            return None

        wait = gates.limiter.acquire(task["host"])
        log.info("pace", wait_seconds=round(wait, 3), source=gates.limiter.last_source)

        response, outcome = await transport.fetch(task["url"])
        log.info("response", outcome=outcome,
                 status=getattr(response, "status_code", None),
                 duration_ms=transport.last_duration_ms,
                 bytes=len(response.content) if response else 0)

        if outcome != "ok":
            log.info("retry_scheduled", next_attempt=task["attempt"] + 1,
                     delay_seconds=round(gates.backoff(task["attempt"]), 2))
            return None

        record = parse_and_validate(response)
        log.info("record_written", natural_key=record["natural_key"],
                 extractor_version=record["extractor_version"],
                 schema_version=record["schema_version"])
        record["trace_id"] = task["trace_id"]        # the identifier reaches the sink
        return record

Six events per fetch is a reasonable density: each corresponds to a decision somebody might later question. Carrying the identifier into the stored record is the step that closes the loop — a suspicious value in the dataset becomes a query for its trace, and the trace is the whole story of how it got there.

4. Query it #

# Every event for one record's fetch, in order.
QUERY_TRACE = {"query": {"term": {"trace_id": "…"}}, "sort": [{"@timestamp": "asc"}]}

# Traces that needed more than three attempts, in the last day.
QUERY_HEAVY_RETRIES = {
    "size": 0,
    "query": {"bool": {"filter": [
        {"term": {"event": "retry_scheduled"}},
        {"range": {"@timestamp": {"gte": "now-1d"}}},
    ]}},
    "aggs": {"by_trace": {
        "terms": {"field": "trace_id", "min_doc_count": 3, "size": 100},
        "aggs": {"host": {"terms": {"field": "host", "size": 1}}},
    }},
}

The second query answers “which fetches were expensive, and on which hosts” — a question that is unanswerable without correlation and immediately actionable with it. A host whose traces routinely take four attempts is one where the crawl is fighting something, and it usually resolves to a pacing problem rather than a target problem.

5. Nest spans only if you need them #

Full distributed tracing adds parent and span identifiers so the crawl’s internal structure is visible. It is worth adopting when a crawl has genuinely deep pipelines — render, extract, enrich, write, each with substructure — and unnecessary when a flat trace with an attempt counter answers the questions you actually ask.

_span = contextvars.ContextVar("span_id", default=None)

@contextmanager
def span(name: str):
    parent = _span.get()
    span_id = uuid.uuid4().hex[:16]
    token = _span.set(span_id)
    log.info("span_start", span=name, span_id=span_id, parent_span_id=parent)
    try:
        yield span_id
    finally:
        log.info("span_end", span=name, span_id=span_id)
        _span.reset(token)

Adopt it when a flat trace stops being enough, not in anticipation. Spans add fields to every event and complexity to every query, and a crawler is a much flatter system than the request-response services tracing was designed for.

Verification & Testing #

def test_trace_id_is_stable_across_retries(captured_logs):
    run_fetch_with_failures(failures=2)
    ids = {e["trace_id"] for e in captured_logs}
    assert len(ids) == 1
    assert [e["attempt"] for e in captured_logs if e["event"] == "dispatch"] == [0, 1, 2]

def test_context_does_not_leak_between_tasks(captured_logs):
    asyncio.run(run_concurrent_fetches(count=50))
    by_trace = group_by(captured_logs, "trace_id")
    for events in by_trace.values():
        assert len({e["url"] for e in events}) == 1      # no cross-contamination

def test_trace_id_reaches_the_record(sink):
    record = run_fetch()
    assert record["trace_id"]
    assert sink.last_written()["trace_id"] == record["trace_id"]

def test_no_secrets_in_correlated_events(captured_logs):
    run_fetch_with_token_in_url()
    assert not any("token=" in json.dumps(e) for e in captured_logs)

The second test is the one that catches the asyncio bug: a context variable used incorrectly, or a thread-local substituted for one, produces events attributed to the wrong trace, and the corruption is invisible until someone tries to read a trace and finds two URLs in it.

Compliance & Operational Guardrails #

• The trace identifier is random and carries no information about the URL, the host, or a person.
• Correlated events are redacted before emission, in the same processor chain.
• Trace retention follows the log’s data class, not a separate longer window.
• The identifier on a stored record is removed or expired alongside the record.
• Every crawl decision that could later be questioned emits an event on the trace.

Common Mistakes #

1. A new identifier per attempt. The retry relationship — the whole point — is lost.
2. Passing the identifier as a parameter. It survives until the next helper function is added.
3. Thread-locals in an async crawler. Traces interleave and events are attributed to the wrong fetch.

Frequently Asked Questions #

Should the trace identifier be sent to the target? #

No. A per-request header identifying your internal trace tells the site nothing useful and adds a fingerprintable value to every request. Identify the crawler with a stable token and a contact URL, as the User-Agent guidance describes; keep internal correlation internal.

How does this relate to the natural key? #

They answer different questions. The natural key identifies the record and is stable across crawls; the trace identifies one fetch and is unique to it. Storing both means you can ask “how did this record get here” and “what happened during that particular fetch” — and a record whose values changed between crawls has a different trace for each observation.

Does trace correlation increase log volume? #

By one field per event, which is negligible. What does increase volume is the temptation to log more once correlation makes logs useful. Keep the event count per fetch bounded and deliberate — six is plenty — and sample operational events by trace identifier so a sampled fetch keeps its whole story rather than fragments of it.

Should the trace identifier be stored on child records too? #

Yes, on everything the fetch produced — parent rows, child rows, quarantined records, evidence bundles. It costs one column and it means any artefact of the pipeline can be traced back to the fetch that produced it, which is exactly the question asked when something looks wrong. Child rows in particular are easy to omit and are usually where the confusing values turn up.

How long should trace-level logs be kept? #

Long enough to investigate a problem discovered from the data, which in practice means at least as long as the interval between data reviews — often two to four weeks. Beyond that, keep the aggregate metrics rather than the individual events. Trace logs inherit the retention class of the data they describe, so a crawl handling personal data keeps its logs on the shorter window rather than the convenient one.

Related guides #

• Structured Logging and Log Shipping — the event schema and redaction these events flow through.
• Shipping Structured Crawl Logs to Elasticsearch — mapping trace_id as a keyword so these queries work.
• Documenting Scraping Authorization and Data Lineage — joining a stored field back to the fetch that produced it.