Celery Redis 后端-使队列中的任务作为项目存在

Question

当前设置：celery 运行在 EC2 节点上的 docker 容器（使用我们的产品代码）上运行，创建和处理任务。我们的 backend/broker 是 Redis，运行在 AWS 的 elasticache 中。

目标：能够在任何给定时间查看队列大小（类似于 flower 的监控），希望通过 AWS CloudWatch，但不是必需的。任务内容不相关，因为我熟悉备份 redis 实例，并且可以使用本地工具解析备份以进行任何需要的分析。短期历史数据是高度首选（CloudWatch 可以追溯到 2 周，并且具有 1 分钟数据点的粒度，这非常好）。

根据我对 Flower 工作方式的了解，由于我们目前拥有的安全性groups/restrictions，Flower 将无法使用。此外，flower 仅在您访问该页面时进行监控，因此不会保存任何历史数据。

Elasticache 已经在 CloudWatch 中内置了 redis 中的项目数。在我看来，这是实现目标的最佳途径。然而，目前队列代表 redis 中的一个项目（无论队列中有多少任务）。这是解析为 json:

的 redis 备份示例

[{
"1.api_data__cached_api_route.000":"{\"i1\": 0, \"i2\": 1, \"i3\": 0}",
"1.api_data__cached_api_route.001":"{\"i1\": 0, \"i2\": 0, \"i3\": 0}",
"1.api_data__cached_api_route.002":"{\"i1\": 1, \"i2\": 1, \"i3\": 0}",
"staging_event_default":["{\"id\":\"b28b056c-1268-4577-af8a-9f1948860502\", \"task\":{...}}, "{\"id\":\"52668c46-3972-457a-be3a-6e27eedd26e3\", \"task\":{...}}]
}]

Cloudwatch 将其视为 4 个项目、3 个缓存 api 路由和 1 个队列。即使队列有数千个项目，它仍会显示为 4 个项目。 #(队列中的项目)和#(队列中的项目和其他缓存的项目)之间的差异很好，因为这个监控工具将主要用于查看队列是否得到了可怕的备份，并且队列的大小会相形见绌缓存的 api 路由数。

要继续沿着这条路线前进，最简单的答案是如果 celery 有一个配置选项来使队列中的每个项目成为它自己的 redis 项目。如果对此有一个简单的修复或配置选项，它似乎是最容易实现的。这是我们当前的芹菜配置选项：

flask_app.config.update(
  CELERY_BROKER_URL=redis_host,
  CELERY_RESULT_BACKEND=redis_host,
  CELERY_QUEUES=queue_manager.queues,
  CELERY_ROUTES=queue_manager.routes,
  CELERY_DEFAULT_QUEUE=queue_manager.default_queue_name,
  CELERY_DEFAULT_EXCHANGE=queue_manager.default_exchange_name)

_celery = celery.Celery(flask_app.import_name,
  backend=flask_app.config['CELERY_RESULT_BACKEND'],
  broker=flask_app.config['CELERY_BROKER_URL'])

opts = {
  'broker_url': redis_host,
  'result_backed': redis_host,
  'task_queues': queue_manager.queues,
  'task_routes': queue_manager.routes,
  'task_default_queue': queue_manager.default_queue_name,
  'task_default_exchange': queue_manager.default_exchange_name,
  'worker_send_task_events': True,

  'task_ignore_result': True,
  'task_store_errors_even_if_ignored': True,
  'result_expires': 3600,

  'worker_redirect_stdouts': False,
  'beat_scheduler': redbeat.RedBeatScheduler,
  'beat_max_loop_interval': 5
}
_celery.conf.update(opts)

我运行遇到的另一个选项是 celery-cloudwatch-logs，这似乎与我想要实现的目标一致，但似乎更旨在查看特定内容每项任务，而不是汇总（但我可能在那里错了）。

如果没有满足目标的 perfect/easy 解决方案，我将研究 forking/building 关闭 celery-cloudwatch 以上传相关信息。我们团队继承了目前存在的大部分代码，我对celery的工作原理有一个基本的了解，但还没有深入。

提前感谢任何人的想法、评论和帮助！

Answer 1

要使用redis 代理查看队列的队列长度，只需在redis 中使用llen。例如，llen celery。

Answer 2

如果有人碰巧遇到它，我会post我在这里做了什么。

我们已经在应用程序的其他地方安装并配置了 boto3 以进行 S3 访问，这使得 post 到 CloudWatch 变得非常容易。

我在 Celery class 中添加了一个方法来使用 redis 模块中的 llen 检查队列的长度：

 @staticmethod
  def check_lengths():
    result = {}
    for q in Celery._queues:
      result[q] = Celery._redis.llen(q)
    return result

然后 post访问 Cloudwatch 也相当容易：

    namespace = "Celery/Queue"
    metrics = []
    for qname, qlen in data.items():
      metric = {}
      metric["MetricName"] = "ItemsInQueue"
      metric["Dimensions"] = [ {"Name": "QueueName", "Value": qname} ]
      metric["Value"] = qlen
      metric["Unit"] = "Count"

      metrics.append(metric)

    self.cw_client.put_metric_data(Namespace=namespace, MetricData=metrics)

然后我最终使用 AWS Lambda 在分钟内向端点发送网络请求，然后 post将上述数据发送到 CloudWatch。

Celery Redis 后端-使队列中的任务作为项目存在

Celery Redis backend- Make tasks in queue exist as items

amazon-web-services

redis

celery

amazon-cloudwatch

amazon-elasticache