Compliant Web Scraping

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.

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 #

  1. 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.
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",
})
  1. 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"
  1. 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 }
      }
    }
  }
}
  1. 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:

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.

Common Mistakes #

  1. 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.
  2. No index template. Elasticsearch then guesses mappings, often typing status as text, which breaks numeric range and aggregation queries you rely on for dashboards.
  3. 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.