Shipping structured crawl logs to Elasticsearch #

Metrics tell you a crawl is unhealthy; structured logs tell you why, and this page covers the specific pipeline that moves JSON crawl events off a worker and into Elasticsearch where you can query them. It is a detail guide under Structured Logging and Log Shipping, part of the Pipeline Storage, Deduplication & Monitoring section. The parent page defines the logging schema and the philosophy of one-event-per-line JSON; here we get concrete about writing those lines to stdout, shipping them with Filebeat or Vector, applying an index template plus an ILM policy so indices roll and expire, redacting personal data before it ever leaves the host, and finally querying the result in Kibana.

Problem Framing #

A single scraper worker is easy to debug by tailing its console. A fleet of them across autoscaling nodes is not — the container that logged the failure may be gone by the time you look. Centralizing logs in Elasticsearch gives you a durable, searchable record: you can pull every 429 from one host in the last hour, or trace a single request_id across retries and proxy switches.

A disk buffer is what turns a search-backend outage into a delay rather than a gap.

Two things make this a compliance-sensitive operation rather than a plumbing exercise. First, crawl logs routinely capture URLs, request headers, and sometimes response snippets — any of which can carry personal data — so redaction has to happen before the event leaves the process. Second, logs retained forever become their own liability; an index lifecycle policy that deletes old data on a schedule is what keeps retention defensible.

Step-by-Step Implementation #

Emit one JSON object per line to stdout. Let the container runtime capture the stream; do not have the app write files or talk to Elasticsearch directly. Redact sensitive fields in a log processor before serialization.

Dynamic mapping will guess text for identifiers and ruin aggregations.

import logging, re, sys
from pythonjsonlogger import jsonlogger

EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")

class RedactionFilter(logging.Filter):
    """Strip PII from log records before they are serialized and shipped."""
    def filter(self, record: logging.LogRecord) -> bool:
        for field in ("url", "message", "user_agent"):
            val = getattr(record, field, None)
            if isinstance(val, str):
                val = EMAIL_RE.sub("[redacted-email]", val)
                val = re.sub(r"([?&](token|key|auth)=)[^&]+", r"\1[redacted]", val)
                setattr(record, field, val)
        return True

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(jsonlogger.JsonFormatter(
    "%(asctime)s %(levelname)s %(name)s %(message)s",
    rename_fields={"asctime": "@timestamp", "levelname": "level"},
))
handler.addFilter(RedactionFilter())

log = logging.getLogger("crawler")
log.addHandler(handler)
log.setLevel(logging.INFO)

# Redaction happens automatically before this line reaches stdout:
log.info("fetch complete", extra={
    "request_id": "c1f3-9a", "host": "example.com",
    "status": 429, "retry_count": 2, "url": "https://example.com/u?token=SECRET",
})

Ship stdout with a lightweight collector. Run Filebeat (or Vector) as a sidecar/daemon that reads container logs and forwards to Elasticsearch. A minimal Filebeat config:

filebeat.inputs:
  - type: container
    paths: ["/var/log/containers/*crawler*.log"]
    json.keys_under_root: true      # promote JSON fields to top level
    json.add_error_key: true

processors:
  - drop_fields:
      fields: ["agent", "ecs", "input"]   # trim Beats metadata noise

output.elasticsearch:
  hosts: ["https://es:9200"]
  index: "crawl-logs-%{+yyyy.MM.dd}"
  # Secrets come from the keystore, never inline in this file:
  username: "${ES_USER}"
  password: "${ES_PASS}"

setup.ilm.enabled: true
setup.ilm.policy_name: "crawl-logs-ilm"

Create an index template so every rolled index gets correct field mappings — status as an integer, @timestamp as a date, request_id as a keyword for exact-match aggregation:

PUT _index_template/crawl-logs
{
  "index_patterns": ["crawl-logs-*"],
  "template": {
    "settings": { "index.lifecycle.name": "crawl-logs-ilm" },
    "mappings": {
      "properties": {
        "@timestamp":  { "type": "date" },
        "request_id":  { "type": "keyword" },
        "host":        { "type": "keyword" },
        "status":      { "type": "integer" },
        "retry_count": { "type": "integer" },
        "url":         { "type": "keyword", "ignore_above": 2048 }
      }
    }
  }
}

Attach an ILM policy that rolls indices by size/age and deletes them after your retention window — the enforcement point for a defensible retention limit:

PUT _ilm/policy/crawl-logs-ilm
{
  "policy": {
    "phases": {
      "hot":    { "actions": { "rollover": { "max_age": "1d", "max_size": "20gb" } } },
      "delete": { "min_age": "30d", "actions": { "delete": {} } }
    }
  }
}

Verification & Testing #

Confirm documents are actually landing and are well-formed. Count today’s index and inspect one event:

Silent bulk rejection is the most common way audit events disappear.

curl -s "https://es:9200/crawl-logs-*/_count" | jq '.count'
curl -s "https://es:9200/crawl-logs-*/_search?size=1" | jq '.hits.hits[0]._source'

In Kibana’s Discover view or the _search API, run a KQL query for the signal you care about — every throttled fetch against one host in the last window:

curl -s "https://es:9200/crawl-logs-*/_search" -H 'Content-Type: application/json' -d '{
  "query": { "bool": { "filter": [
    { "term":  { "status": 429 } },
    { "term":  { "host": "example.com" } },
    { "range": { "@timestamp": { "gte": "now-1h" } } }
  ] } }
}' | jq '.hits.total.value'

Critically, test the redaction path: log a synthetic event containing a fake email and a ?token= URL, then search the index for those substrings — a correct pipeline returns zero hits because the values were scrubbed before shipping.

Compliance & Operational Guardrails #

Redact at the source, in-process, before the event touches disk or the network. Once a raw URL with an embedded email reaches Elasticsearch it has been replicated across shards and snapshots, and after-the-fact deletion is far harder than never writing it.
Let the ILM delete phase enforce your retention promise. A written-down “we keep crawl logs 30 days” claim is only credible if a policy actually expires the indices; this is the audit evidence for a data-minimization review.
Keep Elasticsearch credentials in the Filebeat keystore or a secrets manager, never inline in the shipped config, and restrict the log index to internal access — crawl logs reveal your targets and cadence.

Bulk Rejections Are Silent by Default #

The bulk API returns HTTP 200 even when individual documents fail, with per-item errors buried in the response body. A shipper that checks only the status code loses those documents permanently and reports success — which is how audit events disappear without anyone noticing until they are needed.

from elasticsearch import Elasticsearch, helpers

def ship(client: Elasticsearch, index: str, events: list[dict]) -> tuple[int, list[dict]]:
    """Index a batch, returning (succeeded, permanently failed). Never fails silently."""
    actions = ({"_index": index, "_source": event} for event in events)
    succeeded, failures = helpers.bulk(
        client, actions,
        raise_on_error=False,        # collect failures rather than aborting the batch
        max_retries=3,
        initial_backoff=2,
        request_timeout=60,
    )
    permanent = []
    for failure in failures:
        info = next(iter(failure.values()))
        status = info.get("status")
        if status == 429 or (status is not None and status >= 500):
            requeue(info)            # transient: back onto the buffer
        else:
            permanent.append(info)   # mapping conflict, malformed document: dead-letter
    if permanent:
        dead_letter.extend(permanent)
        metrics.log_ship_permanent_failures.inc(len(permanent))
    return succeeded, permanent

The distinction between transient and permanent failures matters as much here as it does for crawl responses. A 429 from a saturated search backend should be retried after a pause; a mapping conflict never will succeed and belongs in a dead-letter store where someone can fix the template and replay it. Retrying a mapping conflict forever is the other common way a shipping pipeline quietly stops making progress.

Alert on the dead-letter count and on shipper lag — the gap between the newest event on disk and the newest document in the index. Both are cheap, and together they cover the failure modes that a “shipper is running” check does not.

Templates and Lifecycle, Set Before the First Document #

The first document written to a new index determines its mapping if no template matches, and dynamic mapping guesses badly: identifiers become full-text fields, durations become strings, and aggregations that should be instant become impossible without a reindex. Register an index template with explicit mappings before any writer starts, and version the template so a change produces a new index rather than a conflict with an existing one.

Pair it with a lifecycle policy that rolls over by size and age and deletes at the retention boundary defined for the log’s data class. Retention configured in the shipper rather than the search backend is retention that does not exist: the search backend is where the data actually lives, and the retention obligations apply to logs exactly as they apply to curated records.

Common Mistakes #

Multiline stack traces as separate documents. Without json.keys_under_root and proper multiline handling, a traceback fragments into many partial events; emit exceptions as a single serialized JSON field instead.
No index template. Elasticsearch then guesses mappings, often typing status as text, which breaks numeric range and aggregation queries you rely on for dashboards.
Shipping before redacting. Scrubbing in an Elasticsearch ingest pipeline still means the raw value crossed the wire and may sit in the transaction log; do it in the application first.

Frequently Asked Questions #

Filebeat or Vector — which should I use? #

Both read stdout and write to Elasticsearch reliably. Filebeat integrates most tightly with the Elastic Stack (native ILM setup, ECS fields), while Vector offers a richer in-flight transform language and multiple sinks, which helps if you also fan logs out to S3 or Kafka. For an Elasticsearch-only target, Filebeat is the lower-friction choice.

How do I trace one crawl across retries and proxy swaps? #

Attach a stable request_id to the logging context at the start of a fetch and include it in every subsequent event — retries, backoff waits, proxy switches. In Kibana, filtering on that single keyword field reconstructs the full lifecycle of the request in order.

Does redaction handle personal data in the response body? #

Only if you log the body, which you generally should not. Log metadata — status, latency, content-length, request_id — rather than scraped content. If a snippet is unavoidable for debugging, run it through the same redaction filter and truncate it hard before it enters the log record.

Structured Logging and Log Shipping — the parent technique defining the JSON event schema shipped here.
Data Retention and GDPR Deletion Workflows — aligning the ILM delete phase with formal retention obligations.
Implementing Polite Rate Limiting — the upstream behavior whose 429 events these logs let you audit.