Geospatial Portal

Aggregating Celery Logs with Loki

This guide walks through shipping structured Celery worker logs into Loki so ingestion and harvest failures become searchable by queue, worker, and task in seconds rather than by grepping across scattered pod logs.

It is a hands-on companion to Monitoring and Observability for Geospatial Portals and sits within the wider Infrastructure Orchestration & Configuration Management practice; read the parent guide first for why queue depth and task latency are the golden signals of an ingestion pipeline and how logs complement those metrics. This page is narrowly about the log path: emit JSON, label it, store it, and query it.

Prerequisites

  • Celery 5.3+ workers running under a geospatial portal (metadata harvesting, tile seeding, or bulk ingestion tasks).
  • Loki 2.9+ and Promtail or Grafana Alloy deployed in the monitoring namespace; Grafana with Loki configured as a data source.
  • Worker pods writing logs to stdout/stderr (the twelve-factor default) so a node agent can tail them from the container runtime.
  • Ability to set the Celery logger/formatter and to deploy a DaemonSet in the Kubernetes cluster.
  • python-json-logger (or an equivalent JSON formatter) available in the worker image.

The pipeline below traces one task log line from the worker’s stdout, through the node agent that labels it, into Loki, and onto a Grafana logs panel linked to the metrics view.

Celery worker stdout through Promtail labelling into Loki and Grafana LogQL A Celery worker emits a structured JSON log line to stdout carrying the task id, queue, duration, and status. Promtail, running as a DaemonSet on the node, tails the container log file, parses the JSON, and attaches queue and worker labels. It pushes the labelled stream to Loki, which indexes the labels and stores the line. Grafana queries Loki with LogQL to render a logs panel and to compute p95 task duration, linked to the same time window as the metrics dashboard. Celery worker JSON to stdout task_id · queue duration · status Promtail / Alloy DaemonSet · tails file JSON pipeline stage labels: queue, worker status Loki indexes labels stores stream Grafana LogQL logs panel p95 task duration tail push query

Step-by-step implementation

1. Emit structured JSON logs from the workers

A plain human-readable log line forces Promtail to regex-parse fragile text. Configure Celery’s logger to emit one JSON object per line, carrying the fields you will later filter on: task name, task id, queue, duration, and terminal status. Wire it through Celery’s signals so every task is covered uniformly.

import json, logging, time
from celery import Celery
from celery.signals import task_prerun, task_postrun

app = Celery("geoportal")
logger = logging.getLogger("geoportal.tasks")

_started: dict[str, float] = {}

@task_prerun.connect
def _mark_start(task_id=None, task=None, **_):
    _started[task_id] = time.monotonic()

@task_postrun.connect
def _log_result(task_id=None, task=None, state=None, **_):
    duration_ms = round((time.monotonic() - _started.pop(task_id, time.monotonic())) * 1000, 1)
    logger.info(json.dumps({
        "event": "task_complete",
        "task": task.name,
        "task_id": task_id,
        "queue": getattr(task.request, "delivery_info", {}).get("routing_key", "default"),
        "duration_ms": duration_ms,
        "status": state,          # SUCCESS / FAILURE / RETRY
    }))

Emitting to stdout keeps the workers twelve-factor and lets the node agent do all the shipping — no log files to rotate inside the container. This mirrors the codified-configuration approach the ingestion tier already uses in Scaling Celery Pipelines for Bulk Ingestion.

2. Deploy Promtail to tail and label pod logs

Promtail runs as a DaemonSet so every node’s container logs are covered. The pipeline_stages parse the JSON line and promote a small, bounded set of fields to Loki labels — keep this set tiny, because every label value is a separate stream and high-cardinality labels (like task_id) will blow up Loki’s index. Put task_id in the log body, not a label.

server:
  http_listen_port: 9080
clients:
  - url: http://loki.monitoring.svc:3100/loki/api/v1/push
scrape_configs:
  - job_name: celery-pods
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        regex: celery-worker
        action: keep
      - source_labels: [__meta_kubernetes_namespace]
        target_label: namespace
      - source_labels: [__meta_kubernetes_pod_name]
        target_label: pod
    pipeline_stages:
      - docker: {}
      - json:
          expressions:
            queue: queue
            status: status
            duration_ms: duration_ms
      - labels:
          queue:
          status:

3. Confirm Loki is receiving the labelled streams

Loki needs no per-source config for this — it accepts whatever labelled streams Promtail pushes. Confirm the labels arrived so LogQL queries can select on them:

# List label values Loki has seen for the queue label
curl -s "http://loki.monitoring.svc:3100/loki/api/v1/label/queue/values"
#   expect: {"status":"success","data":["ingest","harvest","tiles","default"]}

4. Write LogQL for failures and p95 task duration

LogQL filters by label first (cheap, indexed) then by line content (a scan over the narrowed stream). This query streams every failed task on the ingest queue:

{app="celery-worker", queue="ingest", status="FAILURE"} | json | line_format "{{.task}} {{.task_id}}"

LogQL can also compute metrics from logs. This extracts duration_ms from the JSON and returns the 95th-percentile task duration per queue over five minutes — a log-derived companion to the queue-depth metric on the Celery board:

quantile_over_time(0.95, {app="celery-worker"} | json | unwrap duration_ms [5m]) by (queue)

5. Build a Grafana logs panel linked to metrics

Add a Logs panel with the failure query as its target, and place it on the same dashboard row as the queue-depth and task-latency metric panels so an operator can pivot from a metric spike straight to the offending lines in the same time window. The panel fragment:

{
  "title": "Celery task failures (ingest)",
  "type": "logs",
  "datasource": { "type": "loki", "uid": "loki" },
  "targets": [
    {
      "expr": "{app=\"celery-worker\", queue=\"ingest\", status=\"FAILURE\"} | json",
      "refId": "A"
    }
  ],
  "options": { "showTime": true, "wrapLogMessage": true, "dedupStrategy": "exact" }
}

Labels are the index, and the index must stay small

Log aggregation systems that index by label share the cardinality constraint that metrics have, and ingestion pipelines are the workload most likely to violate it. The tempting labels are the ones that identify the unit of work — dataset id, record identifier, source URL, task id — and every one of them is unbounded. A harvest of fifty thousand records with a per-record label produces fifty thousand streams, each with a handful of lines, and the query that was supposed to find one failure instead times out reading metadata.

The discipline is the same as for metrics but with a friendlier remedy: everything you cannot label, you can still put in the line. A structured log line carrying the dataset id as a field is fully searchable within a stream; it simply is not part of the index. So label by the things that are bounded and that you would use to narrow a search — queue, task name, worker, environment, severity — and put the identifiers in the payload.

What goes in the index and what goes in the line Two columns: index labels holding bounded values, and payload fields holding unbounded identifiers, with the rule for deciding between them. Labels — the index bounded, and you would filter by them queue: harvest · index · publish task: the registered task name worker: pod or hostname env, severity, attempt bucket a few dozen streams in total Payload — the line unbounded, still searchable in a stream record id, dataset id, source URL task id and correlation id exception type and traceback timings, counts, byte sizes one stream, millions of lines, fast to scan

Two practices make the payload half genuinely usable. Emit one line per task completion with all the fields on it rather than several partial lines, so a single match answers the whole question rather than requiring a join across lines. And carry a correlation id from whatever triggered the work — the harvest run, the publish request, the API call — through every task it spawns, because the interesting failure is rarely one task; it is one run in which fourteen tasks out of fifty thousand failed for two different reasons.

Verification 

# 1. A worker actually emits JSON to stdout
kubectl logs -n geoportal deploy/celery-worker --tail=3
#   expect lines like: {"event":"task_complete","task":"...","status":"SUCCESS",...}

# 2. Promtail is healthy and shipping (targets active, no send errors)
kubectl logs -n monitoring ds/promtail --tail=20 | grep -i -e "level=error" -e "batch"

# 3. Loki returns recent failure lines for the ingest queue
curl -s -G "http://loki.monitoring.svc:3100/loki/api/v1/query_range" \
  --data-urlencode 'query={app="celery-worker",queue="ingest",status="FAILURE"}' \
  --data-urlencode 'limit=5' | grep -o '"status":"success"'

# 4. The p95 query returns a numeric series in Grafana Explore
#    quantile_over_time(0.95, {app="celery-worker"} | json | unwrap duration_ms [5m]) by (queue)

A JSON line in step 1 and a "status":"success" with non-empty data in step 3 confirm the full path from worker to Loki. If the queue label is missing in step 3, the JSON pipeline stage failed to parse — check that the log line is valid JSON and not wrapped in extra text.

Retries make the log lie unless you record the attempt

Ingestion pipelines retry, and retries turn a log into a misleading account of what happened. A task that failed twice and succeeded on the third attempt writes two error lines and one success line; counting error lines says the pipeline is failing, counting success lines says it is fine, and neither is the truth. The truth — one unit of work, eventually successful, after two transient failures — is only recoverable if the attempt number and the final outcome are recorded explicitly.

Three log lines, one unit of work A timeline of one task with attempts one and two failing and attempt three succeeding, alongside the terminal record that states the outcome, the attempt count, and the reason for the earlier failures. One record, three lines attempt 1 · failed upstream timeout attempt 2 · failed same timeout attempt 3 · succeeded after backoff terminal record: outcome=success · attempts=3 · last_error=timeout this is the line to count, alert on, and report — the attempt lines are for diagnosis count attempt lines and the pipeline looks broken; count success lines and the instability is invisible

Emit that terminal record from the task’s success and failure handlers so it exists exactly once per unit of work, and build every dashboard and alert on it. The attempt lines remain valuable — a run where the retry rate tripled while the eventual success rate stayed at 100% is the early warning that an upstream is degrading — but they answer a different question and should live on a different panel. Keeping the two separated is what lets a report to a data owner say honestly that every record was ingested, while an engineer sees that it took twice the usual work.

Troubleshooting matrix

Symptom Likely cause Fix
No streams appear in Loki Promtail relabel_configs dropped the pods Confirm the app: celery-worker label matches the keep regex; check kubectl logs ds/promtail
Lines arrive but queue/status labels are empty JSON stage could not parse the line Ensure logs are pure JSON per line; remove any prefix the Celery formatter adds
Loki rejects pushes with per-stream rate limit Too many label combinations (high cardinality) Remove task_id or timestamps from labels:; keep only bounded fields like queue and status
unwrap duration_ms returns no data Field is a string, not a number, or absent Emit duration_ms as a JSON number; confirm the field name matches the json expression
Query is slow / times out Filtering on line content without a label matcher Add {queue="ingest"} before the `
Grafana logs panel shows nothing Data source points at Prometheus, not Loki Set the panel datasource to Loki and re-run the query

Up one level: Monitoring and Observability for Geospatial Portals.