kodhive/leek

Leek

镜像概述与主要用途

Leek是一款专为Celery任务设计的监控工具，其核心优势在于可通过单个容器实例同时监控多个消息代理（Broker），而其他工具通常仅支持单代理监控。此外，Leek提供环境分支管理、多应用支持、Google SSO认证、数据可视化图表、任务异常监控、高级过滤搜索、数据持久化及Slack通知等功能，旨在解决传统Celery监控工具的痛点，提供可靠的监控能力和良好的用户体验，帮助用户快速定位并响应任务问题。

Leek并非任务/worker控制工具（如终止/启动任务、重启worker），也非可安装/导入的软件包，而是以Docker镜像形式发布的完整应用。

核心功能与特性

多代理监控

通过Leek Agent，单个Leek实例可同时连接并监控多个消息代理的Celery任务，无需为每个代理部署独立实例。

多环境支持

为每个消息代理配置环境标签（如qa、stg、prod），任务事件将自动关联环境信息，支持按环境筛选任务数据。

Google SSO认证

支持通过GSuite组织账户或个人Gmail账户登录，提升访问安全性与便捷性。

增强数据存储

任务事件并非存储于易失性内存，而是索引至Elasticsearch，确保数据持久化与快速检索/搜索。

直观用户界面

提供设计精良的UI，相比命令行工具或简陋界面的替代方案，显著提升用户体验。

通知机制

可配置通知规则，当任务满足特定条件（如状态、名称、环境、运行时间阈值）时触发Slack告警，及时响应关键事件。

异常监控

通过异常名称聚合失败任务，统计每种异常的发生次数、恢复状态、待处理数、失败数及严重级别，快速识别系统问题。

数据可视化图表

生成多维度图表，包括任务状态分布、队列分布、Top 5执行任务、Top 5慢任务、任务执行趋势、队列趋势、失败趋势等，直观展示系统状态。

高级过滤与搜索

支持按任务状态、名称、环境、异常类型等多维度过滤，快速定位目标任务。

使用场景与适用范围

• 多环境Celery集群监控：需区分qa、stg、prod等环境的任务数据，统一查看各环境任务状态。
• 多代理管理：存在多个消息代理（如RabbitMQ、Redis）时，通过单实例集中监控。
• 任务数据持久化：需要长期存储任务事件，用于历史分析或问题回溯。
• 异常告警需求：需实时获取关键任务失败、超时等事件通知。
• 可视化监控需求：需要直观图表展示任务执行趋势、队列负载等指标。

使用方法与配置说明

本地演示部署

通过以下步骤快速启动Leek本地演示环境（包含Leek应用、消息代理、Elasticsearch、测试Celery worker及任务发布者）：

RabbitMQ演示

bash
curl -sSL https://raw.githubusercontent.com/kodless/leek/master/demo/docker-compose-rmq-no-auth.yml > docker-compose.yml
docker-compose up

Redis演示

bash
curl -sSL https://raw.githubusercontent.com/kodless/leek/master/demo/docker-compose-redis-no-auth.yml > docker-compose.yml
docker-compose up

服务启动后，访问 http://0.0.0.0:8000，创建与 LEEK_AGENT_SUBSCRIPTIONS 中 app_name 一致的应用（默认示例为 leek）即可开始使用。

Docker Compose配置示例

以下是完整的docker-compose配置示例（以RabbitMQ为例），包含Leek应用、测试worker、发布者、消息代理及Elasticsearch：

yaml
version: "2.4"
services:
  # 主应用
  app:
    image: kodhive/leek
    environment:
      # 日志级别
      - LEEK_API_LOG_LEVEL=INFO
      - LEEK_AGENT_LOG_LEVEL=INFO
      # 组件启用状态
      - LEEK_ENABLE_API=true
      - LEEK_ENABLE_AGENT=true
      - LEEK_ENABLE_WEB=true
      # URL配置
      - LEEK_API_URL=http://0.0.0.0:5000
      - LEEK_WEB_URL=http://0.0.0.0:8000
      - LEEK_ES_URL=http://es01:9200
      # 认证配置（本地演示禁用）
      - LEEK_API_ENABLE_AUTH=false
      # 代理订阅配置（监控目标）
      - |
        LEEK_AGENT_SUBSCRIPTIONS=
        [
          {
            "broker": "amqp://admin:admin@mq//",  # 消息代理地址
            "backend": null,  # 结果后端（可选）
            "exchange": "celeryev",  # Celery事件交换机
            "queue": "leek.fanout",  # 消费队列名
            "routing_key": "#",  # 路由键
            "org_name": "mono",  # 组织名称
            "app_name": "leek",  # 应用名称
            "app_env": "prod",  # 环境标签
            "prefetch_count": 1000,  # 预取数量
            "concurrency_pool_size": 2  # 并发池大小
          }
        ]
      - LEEK_AGENT_API_SECRET=not-secret  # Agent API密钥（演示用）
    ports:
      - 5000:5000  # API端口
      - 8000:8000  # Web UI端口
    depends_on:
      mq:
        condition: service_healthy

  # 测试Worker（仅演示用）
  worker:
    image: kodhive/leek-demo
    environment:
      - BROKER_URL=pyamqp://admin:admin@mq:5672  # 连接测试消息代理
    depends_on:
      mq:
        condition: service_healthy

  # 测试任务发布者（仅演示用）
  publisher:
    image: kodhive/leek-demo
    environment:
      - BROKER_URL=pyamqp://admin:admin@mq:5672
    command: >
      bash -c "python3 publisher.py"  # 执行任务发布脚本
    depends_on:
      mq:
        condition: service_healthy

  # 测试消息代理（RabbitMQ，仅演示用）
  mq:
    image: rabbitmq:3.8.9-management-alpine
    environment:
      - RABBITMQ_DEFAULT_USER=admin
      - RABBITMQ_DEFAULT_PASS=admin
      - "RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=-rabbit log [{console,[{level,error}]}]"
    ports:
      - 15672:15672  # 管理界面端口
      - 5672:5672  # 消息代理端口
    healthcheck:
      test: [ "CMD", "nc", "-z", "localhost", "5672" ]
      interval: 2s
      timeout: 4s
      retries: 20

  # 测试Elasticsearch（仅演示用）
  es01:
    image: elasticsearch:7.10.1
    container_name: es01
    environment:
      - node.name=es01
      - cluster.name=es-docker-cluster
      - cluster.initial_master_nodes=es01
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    command: ["elasticsearch", "-Elogger.level=ERROR"]
    healthcheck:
      test: ["CMD-SHELL", "curl --silent --fail localhost:9200/_cluster/health || exit 1"]
      interval: 30s
      timeout: 30s
      retries: 3
    ulimits:
      memlock:
        soft: -1
        hard: -1
    ports:
      - 9200:9200  # Elasticsearch端口

关键配置参数说明

其中，LEEK_AGENT_SUBSCRIPTIONS 是核心配置，每个订阅项需指定消息代理地址（broker）、环境标签（app_env）、应用名称（app_name）等，以区分不同来源的任务事件。