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.

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" }
}

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.

Troubleshooting matrix #

Related #

• Monitoring and Observability for Geospatial Portals — the metrics-and-logs signal model this log path completes.
• Scaling Celery Pipelines for Bulk Ingestion — the worker fleet whose failures these logs surface.

Up one level: Monitoring and Observability for Geospatial Portals.