Managing Celery

Celery Management Skill

Analyze and manage Celery task queues with safe, read-only operations.

MANDATORY: Two-Phase Execution

You MUST follow this two-phase pattern. Skipping Phase 1 causes hallucinated worker/task names.

Phase 1: Discovery (ALWAYS run first)

#!/bin/bash

# 1. List active workers
celery -A "$CELERY_APP" inspect active_queues

# 2. Worker stats
celery -A "$CELERY_APP" inspect stats

# 3. List registered tasks
celery -A "$CELERY_APP" inspect registered

# 4. Active tasks
celery -A "$CELERY_APP" inspect active

# 5. Flower API (if available)
curl -s "http://${FLOWER_HOST:-localhost}:5555/api/workers" | jq 'keys'

Phase 1 outputs:

• Active workers and their queues
• Registered task names
• Worker stats (pool size, prefetch count)

Phase 2: Analysis (only after Phase 1)

Only reference workers, tasks, and queues confirmed in Phase 1.

Shell Script Patterns

Helper Function

#!/bin/bash

# Core Celery inspect helper — always use this
celery_inspect() {
    local command="$1"
    celery -A "${CELERY_APP}" inspect "$command" --json 2>/dev/null
}

# Celery control helper (read-only commands only)
celery_control() {
    local command="$1"
    celery -A "${CELERY_APP}" control "$command" --json 2>/dev/null
}

# Flower API helper
flower_api() {
    local endpoint="$1"
    curl -s "http://${FLOWER_HOST:-localhost}:${FLOWER_PORT:-5555}/api/$endpoint"
}

# Redis broker queue length (if using Redis)
redis_queue_len() {
    local queue="${1:-celery}"
    redis-cli -h "${REDIS_HOST:-localhost}" -p "${REDIS_PORT:-6379}" LLEN "$queue"
}

Anti-Hallucination Rules

• NEVER reference a task name without confirming via inspect registered
• NEVER reference a worker name without confirming via inspect stats or Flower API
• NEVER assume queue names — always check inspect active_queues
• NEVER guess task IDs — always get from active/reserved inspection or Flower
• NEVER assume broker type — could be Redis, RabbitMQ, or others

Safety Rules

• READ-ONLY ONLY: Use only inspect commands (stats, active, registered, reserved, scheduled, active_queues)
• FORBIDDEN: control commands (shutdown, pool_restart, rate_limit, revoke) without explicit user request
• NEVER revoke tasks without explicit user request
• NEVER change rate limits without explicit user request
• Use Flower API (GET only) for dashboard metrics

Common Operations

Worker Health Overview

#!/bin/bash
echo "=== Active Workers ==="
celery_inspect stats | jq 'to_entries[] | {worker: .key, pool: .value.pool, concurrency: .value.pool."max-concurrency", prefetch_count: .value.prefetch_count, total_tasks: .value.total, uptime: .value.clock}'

echo ""
echo "=== Active Queues ==="
celery_inspect active_queues | jq 'to_entries[] | {worker: .key, queues: [.value[].name]}'

echo ""
echo "=== Registered Tasks ==="
celery_inspect registered | jq 'to_entries[] | {worker: .key, tasks: .value}'

echo ""
echo "=== Worker Ping ==="
celery -A "$CELERY_APP" inspect ping --json 2>/dev/null

Task Monitoring

#!/bin/bash
echo "=== Active Tasks ==="
celery_inspect active | jq 'to_entries[] | {worker: .key, tasks: [.value[] | {id, name, args: (.args | tostring | .[0:50]), time_start, worker_pid}]}'

echo ""
echo "=== Reserved Tasks (prefetched) ==="
celery_inspect reserved | jq 'to_entries[] | {worker: .key, count: (.value | length), tasks: [.value[] | {id, name}]}'

echo ""
echo "=== Scheduled Tasks (ETA/countdown) ==="
celery_inspect scheduled | jq 'to_entries[] | {worker: .key, tasks: [.value[] | {id: .request.id, name: .request.name, eta: .eta}]}'

echo ""
echo "=== Revoked Tasks ==="
celery_inspect revoked | jq 'to_entries[] | {worker: .key, revoked: .value}'

Queue Depth Analysis

#!/bin/bash
echo "=== Queue Lengths (Redis broker) ==="
for QUEUE in $(celery_inspect active_queues | jq -r '.[][][].name' | sort -u); do
    LEN=$(redis_queue_len "$QUEUE" 2>/dev/null || echo "N/A")
    echo "$QUEUE: $LEN messages"
done

echo ""
echo "=== Queue Lengths (RabbitMQ broker) ==="
curl -s -u "${RMQ_USER:-guest}:${RMQ_PASSWORD:-guest}" \
    "http://${RMQ_HOST:-localhost}:15672/api/queues/%2F" 2>/dev/null | \
    jq '.[] | select(.name | test("celery|default")) | {name, messages, consumers}' 2>/dev/null || echo "RabbitMQ not available"

Flower Dashboard

#!/bin/bash
echo "=== Workers (Flower) ==="
flower_api "workers" | jq 'to_entries[] | {name: .key, status: .value.status, active: (.value.active | length), completed: .value.stats.total, concurrency: .value.stats.pool."max-concurrency"}'

echo ""
echo "=== Task Types (Flower) ==="
flower_api "tasks" | jq 'to_entries | group_by(.value.name) | map({name: .[0].value.name, count: length, states: (map(.value.state) | group_by(.) | map({state: .[0], count: length}))})' 2>/dev/null

echo ""
echo "=== Recent Tasks (Flower) ==="
flower_api "tasks?limit=20" | jq 'to_entries | sort_by(-.value.received) | .[0:20] | .[] | {id: .key, name: .value.name, state: .value.state, runtime: .value.runtime, received: .value.received}'

Output Format

Present results as a structured report:

Managing Celery Report
══════════════════════
Resources discovered: [count]

Resource       Status    Key Metric    Issues
──────────────────────────────────────────────
[name]         [ok/warn] [value]       [findings]

Summary: [total] resources | [ok] healthy | [warn] warnings | [crit] critical
Action Items: [list of prioritized findings]

Target ≤50 lines of output. Use tables for multi-resource comparisons.

Counter-Rationalizations

Common Pitfalls

• Prefetch multiplier: High prefetch_count causes uneven task distribution — check worker prefetch settings
• Task serialization: Task args must be serializable — pickle is dangerous, use JSON
• Result backend: Without result backend, task results are lost — check CELERY_RESULT_BACKEND
• Worker memory leaks: Long-running workers can leak memory — check --max-tasks-per-child
• Queue routing: Tasks without explicit routing go to default queue — check task_routes configuration
• Visibility timeout: Redis broker visibility timeout defaults to 1h — long tasks may be redelivered
• Beat scheduling: Celery Beat runs scheduled tasks — ensure only one Beat instance runs