celery-cli-monitor 0.1.2

A lightweight CLI tool for monitoring and managing Celery clusters (Redis broker only)

celery-cli-monitor (cm)

一个轻量级的命令行工具，用于监控和管理使用 Redis 作为 broker 的 Celery 集群。 灵感来自 Flower，但以纯 CLI 形式提供 —— 无需 Web 服务器、无需浏览器、无需开放端口。

为什么需要它？

Flower 很好用，但在生产服务器上，你通常只想快速看一眼类似 top 的 worker 状态概览， 而不想打开浏览器或暴露一个 HTTP 端口。本工具正是为此而生，且依赖极少。

功能特性

• 监控
  • cm workers —— 列出所有 worker，含并发数 / 负载 / 已处理任务数
  • cm workers --name <worker> —— 查看详情（进程池、消费队列、已注册任务、正在执行的任务）
  • cm tasks —— 列出 active / reserved / scheduled / revoked（正在执行 / 已保留 / 已调度 / 已撤销）状态的任务
  • cm tasks --state all —— 统一视图：一次展示各状态任务数（active/reserved/scheduled）+ Redis 队列积压。需要查看详情列表请用具体状态
  • cm task-types —— 列出已注册的任务名称
  • cm queues —— 查看 Redis 队列中的消息堆积数
  • cm queue-list <queue> [-n 10] —— 列出队列中等待执行的任务
  • cm results —— 查看 result backend 中各状态（SUCCESS / FAILURE / RETRY …）的任务数统计
  • cm result-list --state FAILURE —— 列出指定状态的任务（如查看所有失败任务）
  • cm result-detail <task-id> —— 查看单个任务的结果详情，失败任务含完整异常信息和 traceback
• 控制
  • cm shutdown <worker> —— 关停 worker
  • cm pool-restart|pool-grow|pool-shrink <worker> —— 重启 / 扩容 / 缩容进程池
  • cm autoscale <worker> --min 2 --max 10 —— 设置自动伸缩范围
  • cm add-consumer|cancel-consumer <worker> --queue <name> —— 添加 / 取消消费队列
  • cm rate-limit <task> <rate> [--worker <w>] —— 设置任务速率限制
  • cm revoke <task-id> [--terminate] —— 撤销任务（可选终止运行中的任务）
• 执行
  • cm apply <task> --args '[1,2]' --kwargs '{"x":1}' —— 提交任务执行
  • cm apply <task> --send ... —— 无需导入任务代码即可发送
• 实时面板
  • cm watch —— 类似 top 的自动刷新视图

安装

方式一：使用 uv（推荐开发者）

本项目使用 uv 管理依赖，它会自动创建虚拟环境并锁定依赖版本。

# 安装 uv（如尚未安装）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 同步全部依赖（含 rich、redis、pytest 等开发依赖）
uv sync

# 通过 uv 运行命令
uv run cm --help
uv run pytest

方式二：从 PyPI 安装（推荐用户）

pip install celery-cli-monitor

安装时会自动带上 celery、redis、rich 三个依赖，开箱即用，无需额外操作。

方式三：从源码安装（开发者）

git clone https://github.com/manbuheiniu/celery-cli-monitor.git
cd celery-cli-monitor

# 使用 uv（推荐）
uv sync
uv run cm --help

# 或使用 pip（可编辑模式，适合开发）
pip install -e ".[dev]"
cm --help

# 构建分发包
pip install build
python -m build

硬依赖为 celery、redis、rich，安装时自动解析，无需手动安装任何额外依赖。

配置

连接信息自动记忆

首次使用时指定 --broker、--backend、--app 或 --timeout，工具会自动将其保存到当前目录下的 .cm_config.json 文件中。之后再执行命令时就无需重复指定：

# 第一次：指定 --broker，自动保存
cm --broker redis://127.0.0.1:6379/0 workers

# 之后：直接使用，无需再传 --broker
cm workers
cm tasks --state active
cm queues

# 指定新的 --broker 会自动更新保存的值
cm --broker redis://10.0.0.1:6380/1 workers

# --timeout 同样会自动保存（默认 1000ms）
cm --timeout 2000 workers   # 之后所有命令都会使用 2000ms

优先级（从高到低）：

1. 命令行参数 --broker / --backend / --app / --timeout（同时会更新已保存的值）
2. 环境变量 CM_BROKER / CM_BACKEND / CM_APP / CM_TIMEOUT
3. 配置文件 .cm_config.json 中保存的值

查看与清除配置

# 查看已保存的配置
cm config show

# 清除已保存的配置
cm config clear
cm config clear --yes    # 跳过确认

配置文件位置

优先级	位置	说明
1	$CM_CONFIG_FILE	通过环境变量自定义路径
2	./.cm_config.json	默认：当前工作目录

完整参数表

参数	环境变量	说明
--broker	CM_BROKER	Redis broker URL，例如 redis://127.0.0.1:6379/0
--backend	CM_BACKEND	Result backend URL，未设置时默认使用 --broker 的值
--app	CM_APP	Celery 应用模块，例如 myproj.celery
--timeout	CM_TIMEOUT	inspect 超时时间，单位毫秒（默认 1000）

使用示例

# 列出所有 worker
cm --broker redis://127.0.0.1:6379/0 workers

# 通过 Celery 应用连接（同时能看到已注册的任务名）
cm --app myproj.celery workers

# 查看某个 worker 的详情
cm workers --name celery@worker1

# 查看正在执行的任务
cm tasks --state active

# 查看已调度（带 ETA）的任务
cm tasks --state scheduled

# 统一视图：一次看全 active + reserved + scheduled 的任务数 + 队列积压
cm tasks --state all

# 撤销任务（支持短 ID 前缀，自动匹配完整 UUID）
cm revoke 8a4da87b

# 终止正在运行的任务
cm revoke 8a4da87b --terminate

⚠️ revoke 对 reserved 任务可能无效（Celery 已知限制）

Celery 的 revoke 只能阻止任务从 broker 中被取出。如果任务已经被 worker 预取到 本地内存（即 reserved 状态），revoke 可能无法阻止它执行。

这与 worker 的 worker_prefetch_multiplier 配置有关（默认值较大时预取更多任务）。

检测：cm revoke 会在任务处于 reserved 状态时自动发出警告。

解决方案（按推荐顺序）：

1. --terminate：如果任务开始执行则发送信号终止进程
2. cm queue-purge <queue>：清空整个队列（影响该队列所有任务）
3. cm pool-restart <worker>：重启 worker 进程池，清空 reserved 缓存
4. 长期修复：在 Celery 应用中设置 worker_prefetch_multiplier = 1， 让 worker 每次只预取一个任务，从根源上避免此问题

# Celery 应用配置建议
app.conf.worker_prefetch_multiplier = 1
app.conf.task_acks_late = True

# 将 worker 进程池扩容 / 缩减 3 个进程
cm pool-grow celery@worker1 -n 3
cm pool-shrink celery@worker1 -n 3

# ⚠️ pool-shrink 会强制终止进程，正在执行的任务可能丢失！
# 安全缩减前请在 Celery 配置中开启延迟确认：
#   app.conf.task_acks_late = True
#   app.conf.task_reject_on_worker_lost = True
# 如需关停整个 worker（优雅等待任务完成），请使用：
#   cm shutdown celery@worker1

# 查看队列堆积（仅支持 Redis）
cm queues
cm queues --queue video

# 列出队列中等待执行的任务（显示 ID、名称、参数、优先级）
cm queue-list celery
cm queue-list video -n 50    # 最多显示 50 条

# 查看各状态任务数统计（SUCCESS / FAILURE / RETRY …）
cm results

# broker 和 result backend 不在同一个 Redis 时，显式指定 --backend
cm --broker redis://127.0.0.1:6379/0 --backend redis://10.0.0.2:6379/1 results

# 列出所有失败任务
cm result-list --state FAILURE

# 查看某个失败任务的详细错误信息和 traceback
# 支持使用 result-list 中显示的短 ID（前 8 位），也支持完整 UUID
cm result-detail 15e867ae
cm result-detail 15e867ae-d590-47da-8108-19b4abd4ee8a

# 提交任务（需要提供应用代码）
cm apply myapp.tasks.add --args '[1, 2]'

# 无需导入代码即可发送任务
cm apply myapp.tasks.add --args '[1, 2]' --send

# 实时面板
cm watch --interval 2

说明：cm results、cm result-list、cm result-detail 直接从 result backend 读取数据。 Result backend 的地址按以下优先级确定：

1. 命令行 --backend 参数（保存到配置文件，下次自动使用）
2. 环境变量 CM_BACKEND
3. Celery 应用配置中的 result_backend
4. --broker 的值（默认行为：未单独指定时自动使用 broker 地址）

也就是说，如果 broker 和 result backend 使用同一个 Redis（最常见的情况），无需额外配置 --backend。

结果保留时间：Celery 默认只保留 1 天（result_expires = 86400 秒）的任务结果， 过期后会被自动清理。如果发现历史结果数量不变或消失，这是 Celery 的自动过期机制所致。 如需更长的保留时间，在 Celery 配置中调整：

app.conf.result_expires = 7 * 24 * 3600  # 保留 7 天
# 或设为 None 表示永不过期（注意 Redis 内存占用）
# app.conf.result_expires = None

任务名显示：Celery 默认的 result backend 不存储任务名。本工具会自动尝试以下方式补全：

• 失败任务：从 traceback 中解析出任务名（无需额外配置）
• 所有任务：在 Celery 配置中开启 result_extended = True 后，新任务的结果会包含任务名、参数等完整信息

app.conf.result_extended = True

与 Flower 的对比

能力	Flower（Web 版）	cm（CLI 版）
硬依赖	celery, tornado, …	仅 celery
常驻进程	是（服务器）	否（watch 除外）
支持 broker	amqp / redis / …	仅 redis
实时事件	是（EventReceiver）	轮询（watch 模式）
认证 / 多用户	是	不适用（单用户，通过 SSH）
Prometheus 指标	是	未包含

项目结构

celery-cli-monitor/
├── pyproject.toml
├── LICENSE
├── README.md
├── src/
│   └── cli_monitor/
│       ├── __init__.py
│       ├── __main__.py     # argparse 入口，路由所有子命令
│       ├── client.py       # 对 celery.app.control 的薄封装
│       ├── config.py       # 配置持久化（.cm_config.json）
│       ├── broker.py       # Redis 队列深度查询
│       ├── results.py      # Result backend 读取（Redis）
│       ├── display.py      # rich / 纯文本表格渲染
│       ├── utils.py        # 时间 / 格式化辅助函数
│       └── commands/
│           ├── workers.py
│           ├── tasks.py
│           ├── control.py
│           ├── queues.py
│           ├── results.py  # cm results / result-list / result-detail 命令
│           ├── apply.py
│           └── watch.py
└── tests/
    ├── test_config.py
    └── test_unit.py

许可证

MIT