Article

Correlating Logs Across Multiprocessing Workers

Generate a batch identifier once in the parent, pass it into each worker as an argument, and have every worker write its own JSON file named by worker index. Records then carry batch_id, worker and a per-worker seq, which is enough to reassemble the run, attribute a failure to a process, and count events without double-counting. It builds on the Structured Logging for Geospatial CLIs guide, part of the broader CLI Architecture & Design Patterns reference.

Prerequisites

Why the Obvious Approaches Fail

Three arrangements suggest themselves, and two of them break in ways that only appear under load or on another platform.

Three write arrangements for worker logs Shared append relies on atomic writes below the pipe buffer size and is platform-dependent. A queue to the parent is correct but adds a serialisation hop and a single point of contention. One file per worker has no contention and reassembles trivially afterwards. all append to one file no coordination needed atomic only below the pipe buffer size interleaved records on a long line, silently platform-dependent queue to the parent correct on every platform one writer, ordered a pickling hop per record and one contended queue fine at low volume one file per worker no shared resource at all ingest tools read a directory needs a merge step if you want one file at the end the default choice

The left-hand column is the one most projects start with, because it works on a developer’s machine and on Linux with short records. It fails the first time a record exceeds the buffer — which a geospatial record does as soon as someone adds a bounding box and a long source path.

Complete Working Implementation

# gistools/parallel_logging.py
from __future__ import annotations

import itertools
import logging
import os
import uuid
from dataclasses import dataclass
from pathlib import Path

from gistools.logging_setup import SpatialJsonFormatter, get_logger

_seq = itertools.count(1)


@dataclass(frozen=True)
class LogContext:
    """Everything a worker needs to log consistently. Picklable by construction."""
    batch_id: str
    log_dir: Path | None
    level: int = logging.INFO

    @classmethod
    def create(cls, log_dir: Path | None, level: int = logging.INFO) -> "LogContext":
        return cls(batch_id=f"b-{uuid.uuid4().hex[:10]}", log_dir=log_dir, level=level)


def init_worker(ctx: LogContext) -> None:
    """Run once per worker process, via Pool(initializer=...)."""
    worker = os.getpid()
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)
    root.handlers.clear()

    if ctx.log_dir is not None:
        ctx.log_dir.mkdir(parents=True, exist_ok=True)
        handler = logging.FileHandler(ctx.log_dir / f"worker-{worker}.jsonl",
                                      encoding="utf-8")
        handler.setFormatter(SpatialJsonFormatter())
        handler.setLevel(ctx.level)
        root.addHandler(handler)

    global WORKER_LOG
    WORKER_LOG = get_logger("gistools.worker").bind(
        batch_id=ctx.batch_id, worker=worker,
    )
    WORKER_LOG.info("worker_started")


def worker_log():
    """The bound logger for this process, with a fresh sequence number."""
    return WORKER_LOG.bind(seq=next(_seq))

The parent creates the context once and hands it to the pool:

# gistools/commands/warp.py
from multiprocessing import get_context
from pathlib import Path

from gistools.parallel_logging import LogContext, init_worker, worker_log


def warp_tile(args) -> str:
    src, dst, dst_crs = args
    log = worker_log().bind(src=str(src), dst_crs=dst_crs)
    log.info("tile_started")
    try:
        result = do_warp(src, dst, dst_crs)          # the real work
    except Exception:
        log.exception("tile_failed")
        raise
    log.info("tile_finished", extra={"bytes_written": result.size})
    return str(dst)


def run(tiles, log_dir: Path | None, workers: int = 8) -> None:
    ctx = LogContext.create(log_dir)
    parent = get_logger("gistools").bind(batch_id=ctx.batch_id, worker=0)
    parent.info("batch_started", extra={"tiles": len(tiles), "workers": workers})

    mp = get_context("spawn")
    with mp.Pool(workers, initializer=init_worker, initargs=(ctx,)) as pool:
        for _ in pool.imap_unordered(warp_tile, tiles, chunksize=4):
            pass

    parent.info("batch_finished")

Step Annotations

  1. LogContext is a frozen dataclass of plain values. It crosses the process boundary through initargs, so it must pickle — which rules out passing a configured logger, a handler, or an open file object, all of which are tempting and none of which survive the trip.
  2. init_worker runs once per process, not once per task. Pool(initializer=...) is called at worker start-up, so the handler is opened once and reused across every task that worker handles.
  3. The worker identifier is the process id. Using os.getpid() rather than an index avoids needing to allocate and pass one, and it matches what appears in system-level diagnostics if the process is killed.
  4. seq is per worker, not global. A globally-ordered counter would need shared state and lock contention; a per-worker counter is free, and combined with worker it still orders every record within its own process, which is what a traceback reconstruction needs.
  5. The parent uses worker=0 deliberately. Reserving zero for the parent means a query filtering on worker != 0 selects exactly the work records, without a separate flag.

One Named Gotcha: fork Inherits the Parent’s Open Handler

Under the fork start method, a worker inherits the parent’s file descriptors — including the parent’s log handler. Both processes then hold a descriptor for the same file with independent write offsets, and their output overwrites rather than interleaves. The result is a log file with holes in it, which is worse than duplicated records because nothing looks wrong until a query comes up short.

init_worker calls root.handlers.clear() before adding its own, which fixes it. But the more robust answer is not to use fork at all with GDAL in the process, for the separate reasons covered in the guide on multiprocessing with GDAL. Under spawn, a worker starts with no inherited handlers and init_worker is the only thing that installs one.

Reassembling the Run

Once the batch has finished, the per-worker files merge with a sort on timestamp. Because every record carries batch_id, several runs can share a directory without confusion.

From four files to one queryable stream Each worker writes its own newline-delimited JSON file with a monotonic sequence number. A merge sorted by timestamp produces one stream in which batch id groups a run, worker attributes a record, and sequence orders records within a process. worker-4471.jsonl worker-4472.jsonl worker-4473.jsonl worker-4474.jsonl merge, sort by ts no parsing needed batch_id groups the run worker attributes the record seq orders within a process Timestamps from different processes can tie; seq breaks the tie within a worker, and across workers the order genuinely is undefined — which is a fact about the run, not a defect in the log.

That final point saves a lot of wasted effort. People try to impose a global order on records from parallel processes and end up adding a lock to achieve it. The order does not exist; what matters is that each worker’s own story is ordered, and that is free.

Once the fields are in place, three queries answer most of what anyone asks after a parallel run, and each is a one-liner against the merged stream.

What the correlation fields let you ask Grouping by worker reveals an unbalanced pool. Grouping failures by worker reveals a bad node. Filtering by worker and sequence around a failure reconstructs what that process was doing before it broke. group by worker, count events one worker far below the others means the split, not the machine, is unbalanced group failures by worker failures concentrated in one process point at that process, not at the data filter one worker, order by seq reconstructs exactly what that process did in the seconds before it failed

The middle row is the one that repays the effort most often. A failure rate that looks like a data problem in aggregate frequently turns out to be one worker on a node with a full disk, and no aggregate view can distinguish the two.

Verification

Confirm every record carries the three correlation fields, and that the worker set matches the pool size you asked for:

# every record has all three fields
jq -e 'has("batch_id") and has("worker") and has("seq")' logs/*.jsonl > /dev/null \
  && echo "correlation fields present"

# how many distinct workers actually logged
cat logs/*.jsonl | jq -r .worker | sort -u | wc -l

# per-worker sequences are gap-free (a gap means records were lost)
for f in logs/worker-*.jsonl; do
  echo -n "$f "
  jq -s '[.[].seq] | (max - min + 1) == length' "$f"
done

The last check is the one that catches a handler misconfiguration. A gap in a worker’s sequence means records were emitted and not written — usually because the handler level was raised above the record level somewhere, or because a second handler was installed and the first one closed.

Correlating Across Machines, Not Just Processes

The same three fields extend to a distributed batch with one addition: a host identifier. Once workers run on several machines, worker alone is ambiguous — two machines will produce the same process id sooner than you expect — and a failure attributed to “worker 4471” cannot be traced back to a node.

Adding the hostname, or the pod name in a container platform, disambiguates it and costs one field. It also makes the most useful distributed query possible: grouping failures by host, which distinguishes a data problem affecting everyone from a node problem affecting one.

The batch identifier needs no change. It is generated once by whatever submits the work, travels in the task payload, and is the field that ties a run together regardless of how many machines it touched — which is the same role it plays in the task queue guidance.

Shipping the Files Somewhere

Per-worker files solve the write problem and leave a collection problem: the files are on the workers, and the workers are ephemeral. Three arrangements cover almost every deployment.

If the platform captures stdout — a container runtime, a batch service — write the records there instead of to files. The per-worker separation is then handled by the platform, which tags each stream with its container, and no collection step exists.

If the workers share a volume, write to a directory on it named by batch identifier. Collection is then a matter of reading a directory, and a run’s records are already together.

If neither applies, upload the file at the end of each worker’s life, keyed by batch and worker. The risk is a worker killed before the upload, which loses its records — mitigated by uploading periodically rather than only at exit, at the cost of several small objects per worker.

The first is the least work and the most common in practice. The third is where people end up when retrofitting logging onto an existing deployment, and it is worth the effort of moving to one of the other two.

A last detail worth pinning down: the sequence counter must be created inside the worker, not inherited. A module-level itertools.count defined before the pool starts is copied into each child under the fork start method, so every worker begins at the same value and the sequences collide. The counter in the implementation above is created at module import — which under spawn happens fresh in each child, and is another small reason the start method matters.