Logging and Alerting Strategies for Failed CSV Parsing Jobs

13 min read

This page is the telemetry companion to error handling and retry logic for ingestion pipelines within the broader automated permit ingestion and parsing workflows: once a classifier has decided to retry, quarantine, or escalate a failed CSV row, this is the logging and alerting layer that records why and tells the right human when. Municipal permit pipelines ingest CSV exports from legacy records systems, contractor portals, and third-party inspection vendors, and those files almost never conform cleanly to RFC 4180. Parsing failures are therefore routine, and how you log and alert on them is what separates a defensible audit trail from a silent data-loss incident.

What Goes Wrong Without Deliberate Logging

A CSV parse can fail in two fundamentally different ways, and conflating them is the most common operational mistake. A file-level failure — a missing header row, a truncated download, a BOM that shifts every column — means zero records can be trusted from that file. A row-level failure — one malformed date, a ZIP code with a stray quote — means a single application is bad while the other 4,000 in the batch are fine. If your logging cannot tell these apart, on-call engineers get paged for a typo in one contractor’s submission, and a genuinely corrupt county export slips through as a handful of WARNING lines nobody reads.

The compliance stakes are concrete. When a public-records request asks you to account for a permit that “never arrived,” you must be able to reconstruct, from logs alone, that the row was received, why it was rejected, and where it was routed for review. That requires a deterministic correlation_id threaded through every stage and a log record that preserves the offending payload without leaking the applicant’s personal data. The input is messy by nature: mixed encodings, inconsistent quoting, schema drift between annual exports, and the occasional zero-byte file from a cron job that ran before the upstream system finished writing.

How a failed CSV parse is classified, logged, and routed to the right alert channel A single failed CSV row, tagged with a correlation_id, enters a classifier that separates file-level from row-level faults. File-level faults (missing header, BOM, empty file) are logged CRITICAL, halt the batch, and immediately page on-call via PagerDuty or OpsGenie. Row-level faults (a bad date or stray-quote ZIP) are logged WARNING, the row is dead-lettered with its error_code while the batch continues, and the queue feeds a batched shift digest. A row-failure-rate gauge sits below the row-level path: once the rate crosses the row_warn threshold, an escalation path routes around the diagram up to on-call as well. As a cross-cutting side branch, every log call passes through a PII-masking step before reaching a write-once, encrypted log sink. Failed CSV parse → classify, log, route File-level faults page on-call; row-level faults dead-letter and digest — alert on aggregates, never single rows CRITICAL WARNING file-level row-level escalate ≥ row_warn every log call masked write Parse error one failed CSV row + correlation_id Classifier file-level vs row-level fault File-level fault missing header · BOM · empty log CRITICAL · halt batch Page on-call PagerDuty / OpsGenie payload carries correlation_id Row-level fault bad date · stray-quote ZIP log WARNING · batch continues Dead-letter queue recoverable row + error_code → batched shift digest row failure rate ≥ row_warn → page Mask PII redact before sink in processor chain Log sink WORM · encrypted routing & alert log / escalation feed

Step 1: Emit Structured Logs With a Correlation ID

Unstructured print-style logs fail audits and defeat aggregation. Bind a single correlation_id and the file’s identifying metadata at job start, then let every downstream log call inherit that context automatically. structlog makes this binding explicit and produces machine-parseable JSON.

import hashlib
import uuid
from pathlib import Path

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,   # pull in bound context
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),        # one JSON object per line
    ]
)
log = structlog.get_logger()


def begin_csv_job(path: Path, source_system: str, schema_version: str) -> str:
    """Open a parsing job: bind audit context every later log call inherits."""
    correlation_id = str(uuid.uuid4())
    file_hash = hashlib.sha256(path.read_bytes()).hexdigest()
    structlog.contextvars.bind_contextvars(
        correlation_id=correlation_id,
        source_system=source_system,        # e.g. "county-x-accela-export"
        file_hash_sha256=file_hash,         # chain-of-custody anchor
        schema_version=schema_version,      # detect annual export drift
        file_name=path.name,
    )
    log.info("csv.job.started", file_bytes=path.stat().st_size)
    return correlation_id

Passing key-value pairs instead of pre-formatted strings is not a style preference: it prevents log-injection from attacker-controlled CSV cells and guarantees every record parses identically in Elasticsearch, Splunk, or Datadog. The file_hash_sha256 is the same idempotency anchor used by the retry layer, so a quarantined row can be traced back to the exact bytes that produced it.

Step 2: Classify and Log File vs. Row Failures

Give each failure class its own severity, its own machine-readable error code, and its own routing. File-level problems halt the batch; row-level problems are logged and dead-lettered while valid rows continue.

import csv
import io
from typing import Any, Iterator

# Stable codes let clerks filter and let dashboards count by reason.
ERR_MISSING_HEADER = "ERR_MISSING_HEADER"
ERR_ENCODING = "ERR_ENCODING"
ERR_EMPTY_FILE = "ERR_EMPTY_FILE"
ERR_DATE_PARSE = "ERR_DATE_PARSE"
ERR_MISSING_ZIP = "ERR_MISSING_ZIP"
ERR_PERMIT_TYPE_INVALID = "ERR_PERMIT_TYPE_INVALID"

REQUIRED_HEADERS = {"permit_no", "permit_type", "issued_date", "site_zip"}


class FileLevelError(Exception):
    """Structural failure — nothing in the file is trustworthy."""

    def __init__(self, code: str, detail: str) -> None:
        self.code, self.detail = code, detail
        super().__init__(f"{code}: {detail}")


def iter_valid_rows(
    raw: bytes, dead_letter: "DeadLetterQueue"
) -> Iterator[dict[str, Any]]:
    """Yield clean rows; CRITICAL on structural faults, WARNING per bad row."""
    if not raw.strip():
        log.critical("csv.file.failed", error_code=ERR_EMPTY_FILE, byte_offset=0)
        raise FileLevelError(ERR_EMPTY_FILE, "zero-byte or whitespace-only payload")
    try:
        text = raw.decode("utf-8-sig")             # strips a leading BOM if present
    except UnicodeDecodeError as exc:
        log.critical("csv.file.failed", error_code=ERR_ENCODING, byte_offset=exc.start)
        raise FileLevelError(ERR_ENCODING, str(exc)) from exc

    reader = csv.DictReader(io.StringIO(text))
    missing = REQUIRED_HEADERS - set(reader.fieldnames or [])
    if missing:
        log.critical("csv.file.failed", error_code=ERR_MISSING_HEADER,
                     missing_headers=sorted(missing))
        raise FileLevelError(ERR_MISSING_HEADER, f"missing {sorted(missing)}")

    for index, row in enumerate(reader, start=2):   # row 1 is the header
        try:
            yield validate_permit_row(row)          # raises on a bad field
        except RowLevelError as exc:
            log.warning(
                "csv.row.rejected",
                row_index=index,
                error_code=exc.code,
                column=exc.column,
                # Cap the snippet so one giant cell can't bloat the log store.
                raw_snippet=str(row.get(exc.column, ""))[:256],
            )
            dead_letter.put(row_index=index, error_code=exc.code, payload=row)

The dead-letter queue keeps a recoverable row out of the live table without stopping the batch, and the capped raw_snippet gives a clerk enough to fix the record without copying an unbounded payload into the log store. Encoding faults that survive a fallback decode belong to syncing legacy CSV exports to modern databases, where re-decode strategies are covered in depth.

Step 3: Compute a Failure Rate and Route Tiered Alerts

Alert on aggregates, not on every rejected row. Roll the run up into a small summary, then route by severity and by the row-failure rate so a 0.1% typo rate produces a digest while a 40% rate pages an engineer.

from dataclasses import dataclass


@dataclass(slots=True)
class JobSummary:
    correlation_id: str
    source_system: str
    total_rows: int
    rejected_rows: int
    file_error: str | None = None       # set only on a structural failure

    @property
    def failure_rate(self) -> float:
        return self.rejected_rows / self.total_rows if self.total_rows else 1.0


def route_alert(summary: JobSummary, alerts: "AlertSink", row_warn: float = 0.05) -> None:
    """Decide who hears about this job, and how loudly."""
    if summary.file_error is not None:
        # File-level failure: page on-call immediately, the whole batch is dead.
        alerts.page_oncall(
            title=f"CSV file rejected: {summary.file_error}",
            correlation_id=summary.correlation_id,
            source_system=summary.source_system,
        )
        log.critical("alert.paged", reason=summary.file_error)
        return

    if summary.failure_rate >= row_warn:
        # Too many bad rows to be a one-off typo — escalate to engineering.
        alerts.page_oncall(
            title=f"CSV row failure rate {summary.failure_rate:.0%} on "
                  f"{summary.source_system}",
            correlation_id=summary.correlation_id,
            source_system=summary.source_system,
        )
        log.error("alert.escalated", failure_rate=summary.failure_rate)
    elif summary.rejected_rows:
        # Routine bad rows: hand to clerks as a batched shift digest, no page.
        alerts.add_to_digest(summary)
        log.info("alert.digested", rejected_rows=summary.rejected_rows)

Routing file-level failures to a page and routine row rejections to a digest is what keeps notification fatigue from training the team to ignore the channel. The same AlertSink can fan out to PagerDuty, OpsGenie, or a Slack webhook; keep the correlation_id in every payload so the recipient can pull the full structured log without guessing.

Step 4: Suppress Alert Storms During Upstream Outages

When a vendor system breaks, every file it sends fails the same way for an hour. Without suppression, that is one page per cron tick. A short dedup window collapses identical alerts into one, with a periodic reminder rather than a flood.

import time


class AlertDeduplicator:
    """Collapse repeat alerts with the same key inside a cool-down window."""

    def __init__(self, window_s: float = 900.0) -> None:  # 15-minute window
        self._window_s = window_s
        self._last_sent: dict[str, float] = {}

    def should_send(self, dedup_key: str) -> bool:
        """True only once per key per window — e.g. key = source_system+error_code."""
        now = time.monotonic()
        last = self._last_sent.get(dedup_key)
        if last is None or now - last >= self._window_s:
            self._last_sent[dedup_key] = now
            return True
        return False

Key the deduplicator on source_system + error_code so a county whose export is briefly broken yields one page, not forty, while a different failure on a different source still alerts immediately. This pairs with the breaker in building fallback routing for legacy system downtime: when the breaker opens, suppress the per-file alerts and emit a single “source degraded” event instead.

Parameter and Flag Reference

Parameter Type Recommended Rationale for permit CSV jobs
row_warn (failure-rate threshold) float 0.05 Below ~5% rejection is normal typo noise → digest; above it signals schema drift or a bad export → page.
raw_snippet cap int (chars) 256 Enough for a clerk to identify the bad cell; small enough that one runaway field can’t bloat the log store or leak a full record.
dedup window_s float (s) 900 One page per source per failure per 15 min; long enough to ride out a vendor blip, short enough to re-alert a real outage.
log level for rejected rows str WARNING Row rejection is expected and recoverable; reserve ERROR/CRITICAL for batch-stopping faults so dashboards stay meaningful.
correlation_id source str uuid4 per job Stable across all stages so a quarantined row traces back to its file and original bytes.
log retention int (days) 2555 ~7 years to satisfy public-records retention; align to the local mandate and never auto-purge below it.

Common Failure Patterns and Fixes

Per-row alerts drown the on-call channel

Wiring an alert to every csv.row.rejected event is the fastest way to make alerts worthless. Aggregate first: count rejections per job, alert only on the rate crossing row_warn, and send routine rejects to a batched digest. The fix is structural — alert on JobSummary, never on individual RowLevelErrors.

Correlation context is lost across async workers

structlog.contextvars is bound to the current execution context. When you hand rows to a thread pool or an asyncio task group, the bound correlation_id does not always propagate. Re-bind it inside each worker from a value you pass explicitly:

async def process_row(row: dict[str, Any], correlation_id: str) -> None:
    structlog.contextvars.bind_contextvars(correlation_id=correlation_id)
    ...  # every log call in this task now carries the id

Concurrency-heavy loads compound this; the worker-pool plumbing lives in implementing async batch processing for high-volume submissions.

PII leaks into logs

A rejected row often contains an applicant’s name, address, or phone number, and logging it verbatim turns your log store into a regulated data system. Mask before emission, not after:

import re

_PHONE = re.compile(r"\b\d{3}[-.\s]?\d{3}[-.\s]?\d{4}\b")


def mask_pii(value: str) -> str:
    """Redact obvious identifiers before a value reaches the log sink."""
    return _PHONE.sub("[REDACTED-PHONE]", value)

Apply masking inside the logging processor chain so no call site can forget it.

Encoding errors logged as a generic traceback

A UnicodeDecodeError logged as a bare stack trace is invisible to a dashboard counting by error_code. Always attach the stable ERR_ENCODING code and the byte offset, as in Step 2, so the failure is countable and the offset points a clerk at the corrupt region.

Schema drift between annual exports

A county renames issued_date to issue_dt in its new-year export and every row fails ERR_MISSING_HEADER. Logging schema_version and the missing_headers set turns a mystery outage into a one-line diff. Alert on the file-level code immediately — this is never a row-by-row problem.

Audit and Logging Guidance

For a compliance officer to certify a permit’s chain of custody, each parsing job must leave a queryable trail: the job-start record (file hash, source system, schema version, correlation id), every csv.row.rejected event with its error code and dead-letter destination, and the final JobSummary with totals and disposition. Store these under a write-once, read-many policy, encrypt at rest, and gate access with role-based controls so the same logs that satisfy an internal review also withstand a state audit.

Mask or hash all personal and sensitive permit details before emission, retain failure logs for the full period your local public-records law requires (commonly around seven years), and run a monthly job that verifies log completeness and checksum integrity. The dead-letter routing decisions recorded here feed directly into the quarantine-and-replay workflow detailed in the parent error handling and retry logic for ingestion pipelines guide.