Celery Remote Control: Managing Workers Programmatically via CLI and API

In this tutorial, you will learn about Celery Remote Control: Managing Workers Programmatically via CLI and API. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery remote control lets you manage running workers through broadcast commands and the control API, enabling pool resizing, rate limit changes, task inspection, and configuration updates without restarting worker processes.

flowchart LR
    Admin[Admin/CLI] -->|Broadcast Command| B[Broker]
    B --> W1[Worker 1]
    B --> W2[Worker 2]
    B --> W3[Worker 3]
    W1 -->|Reply| Admin
    W2 -->|Reply| Admin
    W3 -->|Reply| Admin
    Admin -->|Control API| App[Celery App]
    App -->|commands| Workers

What You'll Learn

• Using celery control CLI commands
• Using the app.control API
• Pool management (grow/shrink)
• Remote configuration changes
• Task inspection and statistics

Why It Matters

Without remote control, every configuration change requires restarting workers, causing task delays. Remote control lets you adjust pool size, update rate limits, and inspect worker health without interrupting task processing.

Real-World Use

DodaTech's ops team uses remote control to double worker pool size during traffic spikes without restarting. They inspect queue lengths and worker stats through the control API, which feeds their Grafana dashboards.

CLI Control Commands

from celery import Celery

app = Celery('remote', broker='redis://localhost:6379/0')

@app.task
def sample_task(x):
    return x * 2

sample_task.delay(21)
sample_task.delay(42)
sample_task.delay(84)

Control commands:

celery -A remote control status
celery -A remote control inspect active
celery -A remote control inspect stats
celery -A remote control pool_grow 2
celery -A remote control pool_shrink 1

Expected output:

celery@host1: OK
celery@host2: OK
- active: [{'id': 'task-id-1', 'name': 'sample_task'}]
- stats: {'total': 3, 'active': 1, 'processed': 2}
pool_grow: celery@host1: OK (+2 workers, now 6)
pool_shrink: celery@host1: OK (-1 worker, now 5)

Control API in Python

from celery import Celery
import json

app = Celery('remote', broker='redis://localhost:6379/0')

def get_worker_status():
    result = app.control.inspect()
    status = {
        'active_tasks': result.active(),
        'scheduled_tasks': result.scheduled(),
        'reserved_tasks': result.reserved(),
        'stats': result.stats(),
        'registered_tasks': result.registered(),
    }
    print(json.dumps(status, indent=2, default=str))
    return status

def scale_workers(num_additional=2):
    result = app.control.pool_grow(num_additional)
    print(f"Scaled up by {num_additional}: {result}")
    return result

status = get_worker_status()
scale_workers(2)

Expected output:

{
  "active_tasks": {"celery@host": [{"id": "abc", "name": "sample_task"}]},
  "registered_tasks": {"celery@host": ["remote.sample_task"]},
  "stats": {"celery@host": {"total": 150, "active": 1, "prefetch_count": 4}}
}
Scaled up by 2: {'celery@host': {'ok': 'pool will grow'}}

Remote Rate Limit Changes

from celery import Celery

app = Celery('remote', broker='redis://localhost:6379/0')

app.conf.task_default_rate_limit = '10/m'

@app.task(rate_limit='5/m')
def api_call(endpoint):
    result = f"Called {endpoint}"
    print(result)
    return result

app.control.rate_limit('remote.api_call', '20/m')
print("Rate limit updated without restart")

api_call.delay("/users")
api_call.delay("/orders")

Expected output:

Rate limit updated without restart
Called /users
Called /orders
[2026-06-28 10:00:00: INFO] Rate limit for remote.api_call changed to 20/m

Remote Configuration Updates

from celery import Celery

app = Celery('remote', broker='redis://localhost:6379/0')

def update_worker_settings(max_tasks_per_child=None, worker_max_memory=None):
    settings = {}
    if max_tasks_per_child:
        settings['CELERYD_MAX_TASKS_PER_CHILD'] = max_tasks_per_child
    if worker_max_memory:
        settings['CELERYD_MAX_MEMORY_PER_CHILD'] = worker_max_memory

    if settings:
        result = app.control.broadcast('set_worker_settings', arguments=settings)
        print(f"Settings updated: {settings}")
        return result

update_worker_settings(max_tasks_per_child=100, worker_max_memory=500000)

Expected output:

Settings updated: {'CELERYD_MAX_TASKS_PER_CHILD': 100, 'CELERYD_MAX_MEMORY_PER_CHILD': 500000}
[2026-06-28 10:00:00: INFO] Worker settings updated via remote control

Common Mistakes

• Using control commands on the wrong worker -- broadcast commands reach all workers. Use destination=['celery@host1'] to target specific workers. Without destination, every worker applies the command.
• Expecting persistent configuration changes -- remote control changes are in-memory only. Worker restart resets to configured values. Use configuration files or environment variables for permanent changes.
• Rate limit changes not taking effect immediately -- rate limit updates are applied asynchronously. There is a small delay (seconds) between the command and the worker applying the new limit.
• Inspect commands on large clusters -- inspect collects replies from all workers. With 100+ workers, the response can be large and slow. Use destination to query specific workers or query aggregate stats via monitoring.
• Not handling worker unavailability -- if a worker is unreachable, control commands fail silently or time out. Always check the reply dictionary for OK status per worker.

Practice Questions

1. How do you target a specific worker with a control command?
2. What information does inspect.stats() provide?
3. How does control.broadcast differ from control.pool_grow?
4. Why are remote control changes lost on worker restart?
5. How can you check if a control command was successful?

Challenge

Build a remote worker management CLI tool that: (1) lists all workers with their status, pool size, and load, (2) provides commands for pool grow/shrink with size argument, (3) updates rate limits for specific task names, (4) targets commands to specific workers or all workers, (5) logs all actions with timestamps and worker replies, and (6) shows a summary after each command (successful workers, failed workers).

FAQ

Mini Project

Build an auto-scaler using remote control: (1) monitor queue depth every 30 seconds via inspect, (2) if queue depth exceeds threshold, use pool_grow to add workers, (3) if queue depth is near zero for 5 minutes, use pool_shrink to reduce workers, (4) log all scaling actions with queue depth snapshots, (5) implement a cooldown period (60s after any scale action), and (6) send a notification when max_concurrency is reached.

What's Next

Continue with Broadcast Messages to learn how to send commands to all workers. Then explore Worker Inspection for deep worker metrics and statistics.