ryanforgel/notify-server

By ryanforgel

•Updated 10 months ago

Notification forwarding system , from Watchtower to notification channels (Telegram, WeCom)

Image

Developer tools

1.1K

Overview Tags

ryanforgel/notify-server repository overview

⁠通用消息通知服务 (Notification Service)

高性能的通用通知转发系统，基于 Python Quart 异步框架。支持将来自 Watchtower 等服务的通知消息路由到多种通知渠道（Telegram、企业微信等）。

⁠功能特性

多渠道支持: 支持 Telegram、企业微信等多种通知渠道
灵活路由: 基于标签的消息路由规则，支持一对多分发
动态配置: 支持运行时热重载配置，无需重启服务
异步高性能: 基于 Quart 异步框架，支持高并发消息处理
容错设计: 内置重试机制和错误隔离，单个渠道故障不影响其他渠道
插件化架构: 易于扩展新的通知渠道
消息模板: 支持自定义消息模板和变量替换
健康检查: 提供详细的系统健康状态监控

⁠快速开始

⁠安装依赖

pip install -r requirements.txt

⁠配置文件

创建配置文件 dynamic_config/files/d_config.yaml:

receivers:
  telegram_bot:
    type: telegram
    bot_token: "YOUR_BOT_TOKEN"
    chat_id: "YOUR_CHAT_ID"
    template: "🔔 {container_name}: {update_status}\n{content}"
  
  wecom_webhook:
    type: wecom
    webhook_url: "YOUR_WECOM_WEBHOOK_URL"
    msgtype: markdown
    template: "**容器更新通知**\n> {container_name}\n> 状态: {update_status}"

rules:
  - match: watchtower
    receivers:
      - telegram_bot
      - wecom_webhook
  
  - match: critical
    receivers:
      - telegram_bot

⁠启动服务

python app.py

服务将在 http://0.0.0.0:8080 启动。

⁠发送测试消息

curl -X POST http://localhost:8080/notify \
  -H "Content-Type: application/json" \
  -d '{
    "tag": "watchtower",
    "content": "Container updated successfully",
    "container_name": "my-app",
    "update_status": "success"
  }'

⁠API 端点文档

⁠POST /notify

接收并路由通知消息。

请求体:

{
  "tag": "string",           // 必需: 消息标签，用于路由匹配
  "content": "string",       // 必需: 消息内容
  "container_name": "string", // 可选: 容器名称
  "update_status": "string",  // 可选: 更新状态 (success/failure)
  "timestamp": "string",      // 可选: ISO 8601 格式时间戳
  "metadata": {}             // 可选: 额外元数据
}

响应:

200 OK: 消息成功发送到至少一个接收器

{
  "status": "success",
  "results": [
    {
      "receiver": "telegram_bot",
      "success": true,
      "message": "Message sent successfully"
    }
  ]
}

400 Bad Request: 消息格式无效

{
  "error": "ValidationError",
  "message": "Missing required field: tag"
}

404 Not Found: 没有匹配的路由规则

{
  "error": "NoMatchingRule",
  "message": "No routing rule matches tag: unknown"
}

500 Internal Server Error: 所有接收器发送失败

{
  "error": "SendError",
  "message": "Failed to send to all receivers",
  "details": [...]
}

⁠POST /reload

重新加载配置文件。

响应:

200 OK: 配置重载成功

{
  "status": "success",
  "message": "Configuration reloaded successfully",
  "receivers_count": 2,
  "rules_count": 2
}

500 Internal Server Error: 配置重载失败

{
  "error": "ConfigError",
  "message": "Failed to parse configuration file",
  "details": "..."
}

⁠GET /health

检查系统健康状态。

响应:

{
  "status": "healthy",  // healthy | degraded | unhealthy
  "timestamp": "2024-01-01T12:00:00Z",
  "components": {
    "config": {
      "status": "healthy",
      "message": "Configuration loaded successfully"
    },
    "receivers": {
      "telegram_bot": {
        "status": "healthy",
        "message": "Receiver is operational"
      },
      "wecom_webhook": {
        "status": "healthy",
        "message": "Receiver is operational"
      }
    }
  }
}

状态说明:

healthy: 所有组件正常运行
degraded: 部分接收器异常，但服务仍可用
unhealthy: 配置加载失败或关键组件故障

⁠配置文件格式

配置文件使用 YAML 格式，位于 dynamic_config/files/d_config.yaml。

⁠完整配置示例

# 全局配置（可选）
global:
  retry_count: 3        # 重试次数
  retry_interval: 2     # 重试间隔（秒）
  timeout: 10          # 请求超时（秒）

# 接收器配置
receivers:
  # Telegram 接收器
  telegram_bot:
    type: telegram
    bot_token: "1234567890:ABCBAxxxxxxxxx"
    chat_id: "-1001234567890"
    enabled: true
    template: |
      🔔 容器更新通知
      容器: {container_name}
      状态: {update_status}
      时间: {timestamp}
      详情: {content}
  
  # 企业微信接收器 (文本格式)
  wecom_text:
    type: wecom
    webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    msgtype: text
    enabled: true
    template: "容器 {container_name} 更新{update_status}: {content}"
  
  # 企业微信接收器 (Markdown 格式)
  wecom_markdown:
    type: wecom
    webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    msgtype: markdown
    enabled: true
    template: |
      **容器更新通知**
      > 容器名称: {container_name}
      > 更新状态: <font color="info">{update_status}</font>
      > 更新时间: {timestamp}
      
      {content}

# 路由规则
rules:
  # Watchtower 消息路由到所有渠道
  - match: watchtower
    receivers:
      - telegram_bot
      - wecom_markdown
    enabled: true
    priority: 0
  
  # 关键消息只发送到 Telegram
  - match: critical
    receivers:
      - telegram_bot
    enabled: true
    priority: 1
  
  # 测试消息发送到文本格式企业微信
  - match: test
    receivers:
      - wecom_text
    enabled: true
    priority: 0

⁠配置字段说明

⁠Receivers 配置

通用字段:

type: 接收器类型 (telegram | wecom)
enabled: 是否启用 (默认: true)
template: 消息模板，支持变量替换

Telegram 特定字段:

bot_token: Telegram Bot Token
chat_id: 目标聊天 ID（可以是用户 ID、群组 ID 或频道 ID）

企业微信特定字段:

webhook_url: 企业微信机器人 Webhook URL
msgtype: 消息类型 (text | markdown)

⁠Rules 配置

match: 匹配标签（消息的 tag 字段）
receivers: 目标接收器名称列表
enabled: 是否启用规则 (默认: true)
priority: 优先级（暂未使用）

⁠消息模板变量

模板支持以下变量替换：

{tag}: 消息标签
{content}: 消息内容
{container_name}: 容器名称
{update_status}: 更新状态
{timestamp}: 时间戳
{metadata.xxx}: 元数据中的任意字段

⁠部署指南

⁠Docker 部署

⁠1. 构建镜像

docker build -t notification-service:latest .

⁠2. 运行容器

docker run -d \
  --name notification-service \
  -p 8080:8080 \
  -v $(pwd)/dynamic_config/files:/app/dynamic_config/files \
  -e LOG_LEVEL=INFO \
  notification-service:latest

⁠3. 使用 Docker Compose

创建 docker-compose.yml:

version: '3.8'

services:
  notification-service:
    build: .
    container_name: notification-service
    ports:
      - "8080:8080"
    volumes:
      - ./dynamic_config/files:/app/dynamic_config/files:ro
    environment:
      - LOG_LEVEL=INFO
      - CONFIG_PATH=/app/dynamic_config/files/d_config.yaml
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 10s

启动服务：

docker-compose up -d

⁠Kubernetes 部署

⁠1. 创建 ConfigMap

apiVersion: v1
kind: ConfigMap
metadata:
  name: notification-config
  namespace: default
data:
  d_config.yaml: |
    receivers:
      telegram_bot:
        type: telegram
        bot_token: "YOUR_BOT_TOKEN"
        chat_id: "YOUR_CHAT_ID"
    rules:
      - match: watchtower
        receivers:
          - telegram_bot

⁠2. 创建 Secret (用于敏感信息)

apiVersion: v1
kind: Secret
metadata:
  name: notification-secrets
  namespace: default
type: Opaque
stringData:
  telegram-bot-token: "1234567890:ABCdefGHIjklMNOpqrsTUVwxyz"
  wecom-webhook-url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"

⁠3. 创建 Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: notification-service
  namespace: default
spec:
  replicas: 2
  selector:
    matchLabels:
      app: notification-service
  template:
    metadata:
      labels:
        app: notification-service
    spec:
      containers:
      - name: notification-service
        image: notification-service:latest
        ports:
        - containerPort: 8080
          name: http
        env:
        - name: LOG_LEVEL
          value: "INFO"
        volumeMounts:
        - name: config
          mountPath: /app/dynamic_config/files
          readOnly: true
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 30
          timeoutSeconds: 5
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10
          timeoutSeconds: 3
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"
      volumes:
      - name: config
        configMap:
          name: notification-config

⁠4. 创建 Service

apiVersion: v1
kind: Service
metadata:
  name: notification-service
  namespace: default
spec:
  selector:
    app: notification-service
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8080
  type: ClusterIP

⁠5. 部署所有资源

kubectl apply -f configmap.yaml
kubectl apply -f secret.yaml
kubectl apply -f deployment.yaml
kubectl apply -f service.yaml

⁠与 Watchtower 集成

在 Watchtower 容器中配置通知 URL：

docker run -d \
  --name watchtower \
  -v /var/run/docker.sock:/var/run/docker.sock \
  containrrr/watchtower \
  --http-api-update \
  --http-api-metrics \
  --notification-url "http://notification-service:8080/notify" \
  --notification-template '{"tag":"watchtower","content":"{{.Message}}","container_name":"{{.ContainerName}}","update_status":"{{.Status}}"}'

⁠环境变量

支持以下环境变量：

LOG_LEVEL: 日志级别 (DEBUG | INFO | WARNING | ERROR)，默认: INFO
CONFIG_PATH: 配置文件路径，默认: dynamic_config/files/d_config.yaml
HOST: 监听地址，默认: 0.0.0.0
PORT: 监听端口，默认: 8080

⁠开发指南

⁠运行测试

# 运行所有测试
pytest

# 运行单元测试
pytest tests/test_*.py -v

# 运行属性测试
pytest tests/test_*_properties.py -v

# 查看测试覆盖率
pytest --cov=. --cov-report=html

⁠添加新的接收器类型

在 receivers/ 目录创建新的接收器类：

from receivers.base import Receiver
from models.result import SendResult

class MyReceiver(Receiver):
    def __init__(self, config):
        self.config = config
    
    async def send(self, message: str, **kwargs) -> SendResult:
        # 实现发送逻辑
        pass
    
    async def initialize(self) -> bool:
        # 实现初始化逻辑
        return True
    
    async def health_check(self) -> bool:
        # 实现健康检查逻辑
        return True

在 receivers/factory.py 中注册新类型：

from receivers.my_receiver import MyReceiver

factory.register_receiver_type('mytype', MyReceiver)

在配置文件中使用：

receivers:
  my_receiver:
    type: mytype
    # 其他配置...

⁠故障排查

⁠消息未发送

检查日志输出：
```
docker logs notification-service
```

验证配置文件格式：

python -c "import yaml; yaml.safe_load(open('dynamic_config/files/d_config.yaml'))"

测试接收器连接：
```
curl http://localhost:8080/health
```

⁠配置重载失败

检查配置文件语法
查看错误日志中的详细信息
确保配置文件路径正确且有读取权限

⁠性能问题

增加并发处理能力（调整 worker 数量）
检查接收器 API 响应时间
启用 DEBUG 日志查看详细处理流程

Tag summary

Recent tags

Content type

Image

Digest

sha256:cf659648d…

Size

63.7 MB

Last updated

10 months ago

docker pull ryanforgel/notify-server