#Structured Logging and Metrics for Migration Jobs

For a geospatial migration job, structured logging means emitting each log line as a JSON object with stable keys and a correlation ID, and recording a small set of low-cardinality metrics — rows converted, dropped geometries, retries, and duration — through an exporter to CloudWatch, Prometheus, or OpenTelemetry. The correlation ID is the linchpin: generate one per job, bind it once, and carry it on every line and span so a single failed dataset can be reconstructed end to end. Metrics answer “how is the fleet doing” while logs answer “what happened to this dataset”, and the two must share the same identifiers to be useful together. This page shows the exact field schema, a production logging setup, and how to route the signals to a backend. It is the logging and metrics companion to Cost & Observability for Conversion Pipelines.

#Quick-Reference Table

#Why Structure and Correlation Matter

A migration job that logs print("done converting") gives you nothing to query. When ten thousand such lines interleave across concurrent workers, you cannot tell which dataset a message belongs to, cannot filter by severity, and cannot compute how many jobs dropped geometries last night. Structured logging fixes this by making every line a JSON object with predictable keys — timestamp, level, event, job_id, dataset_path, and event-specific fields. Machine-parseable lines are queryable: CloudWatch Logs Insights, Loki, or a lake query can answer “show every job that dropped more than zero geometries in the last hour” in one expression, which is impossible against free text.

The correlation ID is what elevates structured logs from searchable to reconstructable. Generate a single job ID at the start of each migration and bind it to the logger so it attaches automatically to every subsequent line. When a dataset fails, you filter on that one ID and see the entire causal chain — start, each stage, the retry, the error, the terminal record — with nothing from other jobs mixed in. This is the same identifier that ties logs to the metrics and traces described in the parent cost and observability topic, and it is what makes Retry Logic for Cloud Migration Pipelines debuggable: a retried attempt shares the parent job ID but records its own attempt number, so you can see how many times a chunk was tried before it succeeded or was routed to a dead-letter path.

Metrics play the complementary role. Where logs are per-event and high-cardinality, metrics are pre-aggregated and must stay low-cardinality to survive at fleet scale. The discipline that trips up most teams is label hygiene: it is tempting to label a Prometheus counter with the dataset path so you can slice throughput per file, but that path is effectively unbounded and will blow up the time-series database. Keep the identifying detail in the log record and restrict metric labels to a small, bounded set — status, stage, source format — so counters and histograms stay cheap to store and fast to query.

#Structured Logging Implementation

The setup below configures structlog to render JSON, binds a correlation ID per job, and records the key metrics as it goes. It emits a start record, a terminal record on every exit path, and increments counters that an OpenTelemetry or Prometheus exporter can scrape or push.

# structlog>=24.1, python>=3.10
from __future__ import annotations

import time
import uuid
from contextlib import contextmanager
from typing import Iterator

import structlog

# --- Configure JSON rendering once at process start ---
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ]
)
log = structlog.get_logger()

# In-process metric accumulators; flushed by an OTel/Prometheus exporter.
METRICS: dict[str, float] = {
    "rows_converted": 0,
    "geometries_dropped": 0,
    "job_retries": 0,
}


@contextmanager
def migration_job(dataset_path: str, source_format: str) -> Iterator[str]:
    """Bind a correlation ID and emit start/terminal structured records.

    Yields the generated job_id so callers can tag metrics and spans.
    Guarantees a terminal record with duration on success and failure.
    """
    job_id = uuid.uuid4().hex
    structlog.contextvars.bind_contextvars(
        job_id=job_id, dataset_path=dataset_path, source_format=source_format
    )
    start = time.perf_counter()
    log.info("migration.start")
    status = "ok"
    try:
        yield job_id
    except Exception as exc:  # noqa: BLE001 — terminal record must always emit
        status = "failed"
        log.error("migration.error", error=str(exc), exc_type=type(exc).__name__)
        raise
    finally:
        duration = round(time.perf_counter() - start, 4)
        log.info(
            "migration.done",
            status=status,
            duration_seconds=duration,
            rows_converted=METRICS["rows_converted"],
            geometries_dropped=METRICS["geometries_dropped"],
            job_retries=METRICS["job_retries"],
        )
        structlog.contextvars.clear_contextvars()


if __name__ == "__main__":
    with migration_job("s3://bucket/parcels_2026.shp", "shapefile"):
        # ... real conversion would update METRICS as it runs ...
        METRICS["rows_converted"] = 812_004
        METRICS["geometries_dropped"] = 4

#Validation

Run the snippet and confirm the emitted lines are valid JSON sharing one job_id. A start and terminal record for a healthy job should look like this (formatting added for readability):

{"event": "migration.start", "job_id": "9f2c...", "dataset_path": "s3://bucket/parcels_2026.shp", "source_format": "shapefile", "level": "info", "timestamp": "2026-07-13T09:14:02Z"}
{"event": "migration.done", "job_id": "9f2c...", "status": "ok", "duration_seconds": 0.0512, "rows_converted": 812004, "geometries_dropped": 4, "level": "info", "timestamp": "2026-07-13T09:14:02Z"}

Pipe the output through python -m json.tool or jq . to confirm every line parses. Then verify correlation by filtering on the job_id: in CloudWatch Logs Insights, fields @message | filter job_id = "9f2c..." should return exactly the records for that one execution and nothing else. If a line is missing its job_id, it was emitted outside the bound context — a sign that some logging happens before bind_contextvars or after clear_contextvars, which is the most common structured-logging bug.

#Edge Cases and Caveats

Concurrent workers sharing a process. structlog.contextvars is bound per execution context, so async tasks or threads each need their own bind/clear cycle. If you bind once globally and run jobs concurrently in the same process, correlation IDs bleed between jobs and the logs become unreadable. Wrap each unit of work in its own migration_job context, and prefer per-job contexts over a single shared logger for the whole worker.

Retry attempts under one job. When a chunk retries, keep the parent job_id but add an attempt field so each try is distinguishable while still grouping under the dataset. Increment the job_retries counter only on genuine retries, not on the first attempt, or your retry-storm signal will be permanently offset. The retry-classification detail lives in Retry Logic for Cloud Migration Pipelines.

High-cardinality metric labels. Never promote job_id, dataset_path, or geometry counts to metric labels. Doing so creates one time series per unique value and can exhaust the metrics backend within a single large batch. Keep those in logs, and if you need per-dataset cost or throughput, derive it from the log-and-record ledger rather than from labelled metrics.

#Frequently Asked Questions

Why use a correlation ID instead of just the dataset name in logs?

A dataset name is not unique across runs — the same Shapefile is converted many times, and retries reprocess it within a single run. A correlation ID identifies one specific execution, so you can isolate the log lines, metrics, and trace spans for exactly the attempt that failed without pulling in every historical conversion of the same file. Keep both: log the dataset name as an attribute and the job ID as the correlation key.

What log fields should never become metric labels?

Anything high-cardinality: the correlation ID, the dataset path, geometry counts, and timestamps. Using these as Prometheus labels causes a cardinality explosion that can overwhelm the metrics backend. Keep them in the log record, where high cardinality is expected, and restrict metric labels to a small bounded set such as status, stage, and source format.

#Related

• Cost & Observability for Conversion Pipelines — parent topic: signals, dashboards, and SLOs for conversion jobs
• Cost-per-GB Tracking for Conversion Pipelines — turning these metrics into per-dataset cost records
• Retry Logic for Cloud Migration Pipelines — how retries appear in logs and the retry counter
• Fallback Routing for Failed Migration Jobs — routing failures that these logs surface
• Building Batch Conversion Pipelines with Python — the job structure being instrumented

← Back to Cost & Observability for Conversion Pipelines