第327篇：ChatOps：用飞书/钉钉机器人操作网络

关键词：ChatOps、飞书机器人、钉钉机器人、Webhook、交互式运维、对话式操作、消息推送、网络运维

一、引言

ChatOps 的核心思想是：将运维操作从"登录服务器执行命令"转变为"在聊天工具中通过对话完成"。运维人员在飞书或钉钉群聊中发送一条消息，机器人自动执行对应的运维任务并返回结果。这种方式降低了运维操作的门槛，提高了操作的可追溯性和协作效率。

在网络运维场景中，ChatOps 的典型应用包括：

• 一键查询设备状态（BGP 会话、接口利用率、CPU/内存）
• 批量执行配置命令
• 变更结果实时通知
• 告警消息智能推送与确认
• 巡检报告定时生成与推送

二、飞书机器人架构设计

2.1 整体架构

2.2 飞书自定义机器人（Webhook 模式）

飞书自定义机器人通过 Webhook URL 接收消息推送：

# bot/feishu_bot.py
import requests
import json

class FeishuBot:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url

    def send_text(self, text):
        """发送纯文本消息"""
        payload = {
            "msg_type": "text",
            "content": {
                "text": text
            }
        }
        return self._send(payload)

    def send_rich_text(self, elements):
        """发送富文本消息"""
        payload = {
            "msg_type": "interactive",
            "card": {
                "config": {"wide_screen_mode": True},
                "header": {
                    "title": {"tag": "plain_text", "content": "网络运维报告"},
                    "template": "blue"
                },
                "elements": elements
            }
        }
        return self._send(payload)

    def send_markdown(self, title, content):
        """发送 Markdown 消息"""
        payload = {
            "msg_type": "interactive",
            "card": {
                "elements": [
                    {
                        "tag": "markdown",
                        "content": content
                    }
                ],
                "header": {
                    "title": {"tag": "plain_text", "content": title}
                }
            }
        }
        return self._send(payload)

    def _send(self, payload):
        """发送消息到飞书"""
        try:
            resp = requests.post(self.webhook_url,
                               json=payload,
                               timeout=10)
            result = resp.json()
            if result.get('code') != 0:
                print(f"发送失败: {result}")
            return result
        except Exception as e:
            print(f"发送异常: {e}")
            return None

2.3 交互式应用机器人（事件订阅模式）

如需接收用户消息并回复，需创建飞书开放平台应用：

# bot/command_router.py
import re
from flask import Flask, request, jsonify

app = Flask(__name__)

# 命令注册表
COMMANDS = {}

def register(pattern, handler):
    """注册命令装饰器"""
    def decorator(func):
        COMMANDS[re.compile(pattern)] = func
        return func
    return decorator

@register(r'^/status\s+(\S+)')
def cmd_status(device_name):
    """查询设备状态"""
    return execute_status_query(device_name)

@register(r'^/bgp\s+(\S+)')
def cmd_bgp(device_name):
    """查询 BGP 状态"""
    return execute_bgp_query(device_name)

@register(r'^/ping\s+(\S+)\s+(\S+)')
def cmd_ping(src_device, dst_ip):
    """执行 Ping 测试"""
    return execute_ping(src_device, dst_ip)

@register(r'^/interface\s+(\S+)')
def cmd_interface(device_name):
    """查询接口状态"""
    return execute_interface_query(device_name)

@register(r'^/backup\s+(\S+)')
def cmd_backup(device_name):
    """备份设备配置"""
    return execute_backup(device_name)

@app.route('/webhook/feishu', methods=['POST'])
def feishu_webhook():
    """飞书事件回调"""
    data = request.json

    # 获取用户消息文本
    if data.get('event', {}).get('message'):
        text = data['event']['message'].get('text', '')
        sender = data['event']['sender']['sender_id']['open_id']

        # 路由到对应命令
        for pattern, handler in COMMANDS.items():
            match = pattern.match(text)
            if match:
                args = match.groups()
                try:
                    result = handler(*args)
                    # 构造回复消息
                    reply = format_reply(result)
                    return send_reply(data, reply, sender)
                except Exception as e:
                    return send_reply(data, f"执行失败: {str(e)}", sender)

        # 未知命令
        return send_reply(data,
            "未知命令。可用命令：\n"
            "/status <设备名> - 查询设备状态\n"
            "/bgp <设备名> - 查询 BGP 状态\n"
            "/ping <源设备> <目标IP> - Ping 测试\n"
            "/interface <设备名> - 接口状态\n"
            "/backup <设备名> - 备份配置",
            sender)

    return jsonify({"code": 0})

三、钉钉机器人实现

钉钉机器人的实现与飞书类似，但 API 格式不同：

# bot/dingtalk_bot.py
import requests
import json
import hashlib
import hmac
import base64
import time

class DingTalkBot:
    def __init__(self, webhook_url, secret=None):
        self.webhook_url = webhook_url
        self.secret = secret

    def _sign(self):
        """钉钉签名验证"""
        if not self.secret:
            return self.webhook_url

        timestamp = str(round(time.time() * 1000))
        string_to_sign = f'{timestamp}\n{self.secret}'
        hmac_code = hmac.new(
            self.secret.encode('utf-8'),
            string_to_sign.encode('utf-8'),
            digestmod=hashlib.sha256
        ).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return f"{self.webhook_url}&timestamp={timestamp}&sign={sign}"

    def send_text(self, content, at_mobiles=None):
        """发送文本消息"""
        payload = {
            "msgtype": "text",
            "text": {"content": content},
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": False
            }
        }
        resp = requests.post(self._sign(), json=payload)
        return resp.json()

    def send_markdown(self, title, text, at_mobiles=None):
        """发送 Markdown 消息"""
        payload = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            },
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": False
            }
        }
        resp = requests.post(self._sign(), json=payload)
        return resp.json()

    def send_action_card(self, title, content, buttons):
        """发送 ActionCard（含按钮的卡片消息）"""
        btns = []
        for btn in buttons:
            btns.append({
                "title": btn['title'],
                "actionURL": btn['url'],
                "backgroundColor": btn.get('color', '#1F7BCC')
            })

        payload = {
            "msgtype": "actionCard",
            "actionCard": {
                "title": title,
                "text": content,
                "btnOrientation": "0",
                "btns": btns
            }
        }
        resp = requests.post(self._sign(), json=payload)
        return resp.json()

四、告警推送与变更通知

4.1 告警智能推送

# bot/alert_pusher.py
from bot.feishu_bot import FeishuBot

class AlertPusher:
    """告警推送器"""

    ALERT_TEMPLATES = {
        'critical': {
            'title': '🚨 严重告警',
            'color': 'red',
            'at_all': True
        },
        'major': {
            'title': '⚠️ 主要告警',
            'color': 'orange',
            'at_all': False
        },
        'warning': {
            'title': '💡 警告',
            'color': 'yellow',
            'at_all': False
        }
    }

    def push_alert(self, alert):
        """推送告警"""
        severity = alert.get('severity', 'warning')
        template = self.ALERT_TEMPLATES.get(severity,
                                            self.ALERT_TEMPLATES['warning'])

        content = (
            f"**设备**: {alert['device']}\n"
            f"**类型**: {alert['type']}\n"
            f"**描述**: {alert['description']}\n"
            f"**时间**: {alert.get('timestamp', 'N/A')}\n"
            f"**建议**: {alert.get('suggestion', '请检查设备状态')}"
        )

        self.bot.send_markdown(template['title'], content)

    def push_change_notification(self, change):
        """推送变更通知"""
        status_icon = {
            'success': '✅',
            'failed': '❌',
            'in_progress': '🔄'
        }.get(change['status'], '📋')

        content = (
            f"**变更ID**: {change['id']}\n"
            f"**标题**: {change['title']}\n"
            f"**状态**: {status_icon} {change['status']}\n"
            f"**设备**: {', '.join(change['devices'])}\n"
            f"**操作人**: {change['requestor']}\n"
            f"**详情**: {change.get('detail', '')}"
        )

        self.bot.send_markdown(
            f"📋 变更通知 [{change['status']}]",
            content
        )

4.2 定时巡检报告推送

# bot/scheduled_report.py
from apscheduler.schedulers.background import BackgroundScheduler

def generate_daily_report():
    """生成每日巡检报告"""
    devices = load_device_list()
    report_lines = ["## 📊 每日网络巡检报告", ""]
    report_lines.append(f"| 设备 | CPU | 内存 | BGP会话 | 接口Down |")
    report_lines.append(f"|------|-----|------|---------|----------|")

    for dev in devices:
        stats = get_device_stats(dev)
        report_lines.append(
            f"| {dev['name']} | {stats['cpu']}% | "
            f"{stats['memory']}% | {stats['bgp_sessions']} | "
            f"{stats['down_interfaces']} |"
        )

    return '\n'.join(report_lines)

# 每天上午 8:30 推送巡检报告
scheduler = BackgroundScheduler()
scheduler.add_job(
    func=lambda: AlertPusher().bot.send_markdown(
        "每日巡检报告", generate_daily_report()),
    trigger='cron',
    hour=8, minute=30
)
scheduler.start()

五、容器化网络运维平台：Docker/K8s 集成

5.1 Docker 容器编排配置

# Dockerfile - 网络运维工具容器
FROM python:3.11-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc libffi-dev libssl-dev && \
    rm -rf /var/lib/apt/lists/*

# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY bot/ ./bot/
COPY tools/ ./tools/
COPY config/ ./config/

# 暴露 Webhook 端口
EXPOSE 8080

# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8080/health')"

# 启动服务
CMD ["uvicorn", "bot.app:app", "--host", "0.0.0.0", "--port", "8080"]

# docker-compose.yml
version: '3.8'

services:
  # ChatOps 后端服务
  chatops-bot:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8080:8080"
    environment:
      - FLYBOOK_WEBHOOK_URL=${FLYBOOK_WEBHOOK_URL}
      - DINGTALK_WEBHOOK_URL=${DINGTALK_WEBHOOK_URL}
      - DB_CONNECTION=postgresql://admin:password@postgres:5432/chatops
      - REDIS_URL=redis://redis:6379/0
    volumes:
      - ./config:/app/config:ro
      - ./logs:/app/logs
      - ./backups:/app/backups
    depends_on:
      - postgres
      - redis
    restart: unless-stopped
    networks:
      - chatops-net

  # 数据库
  postgres:
    image: postgres:15-alpine
    environment:
      - POSTGRES_DB=chatops
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=${DB_PASSWORD:-password}
    volumes:
      - pg_data:/var/lib/postgresql/data
      - ./scripts/init_db.sql:/docker-entrypoint-initdb.d/init.sql
    ports:
      - "5432:5432"
    networks:
      - chatops-net

  # 缓存与任务队列
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes
    volumes:
      - redis_data:/data
    ports:
      - "6379:6379"
    networks:
      - chatops-net

  # Prometheus 监控
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
      - prom_data:/prometheus
    ports:
      - "9090:9090"
    networks:
      - chatops-net

  # Grafana 可视化
  grafana:
    image: grafana/grafana:latest
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD:-admin}
    volumes:
      - grafana_data:/var/lib/grafana
      - ./monitoring/grafana/dashboards:/etc/grafana/provisioning/dashboards
    ports:
      - "3000:3000"
    depends_on:
      - prometheus
    networks:
      - chatops-net

  # 采集 Worker（后台任务）
  collector-worker:
    build:
      context: .
      dockerfile: Dockerfile
    command: python -m scripts.scheduler
    environment:
      - DB_CONNECTION=postgresql://admin:password@postgres:5432/chatops
      - REDIS_URL=redis://redis:6379/0
    volumes:
      - ./inventory:/app/inventory:ro
      - ./output:/app/output
    depends_on:
      - postgres
      - redis
    restart: unless-stopped
    networks:
      - chatops-net

volumes:
  pg_data:
  redis_data:
  prom_data:
  grafana_data:

networks:
  chatops-net:
    driver: bridge

5.2 K8s 网络策略与部署

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: chatops-bot
  namespace: network-ops
  labels:
    app: chatops-bot
    tier: backend
spec:
  replicas: 2
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: chatops-bot
  template:
    metadata:
      labels:
        app: chatops-bot
        tier: backend
    spec:
      serviceAccountName: chatops-bot
      containers:
        - name: chatops-bot
          image: registry.internal/chatops-bot:v1.2.0
          ports:
            - containerPort: 8080
              name: http
          env:
            - name: FLYBOOK_WEBHOOK_URL
              valueFrom:
                secretKeyRef:
                  name: chatops-secrets
                  key: feishu-webhook
            - name: DB_CONNECTION
              valueFrom:
                secretKeyRef:
                  name: chatops-secrets
                  key: db-connection
          resources:
            requests:
              cpu: 100m
              memory: 256Mi
            limits:
              cpu: 500m
              memory: 512Mi
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 15
            periodSeconds: 20
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
          volumeMounts:
            - name: config-volume
              mountPath: /app/config
              readOnly: true
            - name: log-volume
              mountPath: /app/logs
      volumes:
        - name: config-volume
          configMap:
            name: chatops-config
        - name: log-volume
          emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
  name: chatops-bot
  namespace: network-ops
spec:
  selector:
    app: chatops-bot
  ports:
    - port: 80
      targetPort: 8080
      name: http
  type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: chatops-network-policy
  namespace: network-ops
spec:
  podSelector:
    matchLabels:
      app: chatops-bot
  policyTypes:
    - Ingress
    - Egress
  ingress:
    # 只允许 Ingress Controller 访问
    - from:
        - namespaceSelector:
            matchLabels:
              name: ingress-nginx
      ports:
        - port: 8080
  egress:
    # 允许访问数据库
    - to:
        - podSelector:
            matchLabels:
              app: postgres
      ports:
        - port: 5432
    # 允许访问 Redis
    - to:
        - podSelector:
            matchLabels:
              app: redis
      ports:
        - port: 6379
    # 允许访问飞书/钉钉 API
    - to:
        - ipBlock:
            cidr: 0.0.0.0/0
      ports:
        - port: 443
    # 允许 DNS 解析
    - to:
        - namespaceSelector: {}
      ports:
        - port: 53
          protocol: UDP
---
# ConfigMap: 配置文件
apiVersion: v1
kind: ConfigMap
metadata:
  name: chatops-config
  namespace: network-ops
data:
  scheduler_config.yaml: |
    tasks:
      - name: "每小时采集"
        trigger: cron
        cron: { hour: "*" }
        devices_file: /app/inventory/devices.yaml
        max_workers: 20

5.3 容器日志采集方案

# monitoring/filebeat.yml - Filebeat 日志采集配置
filebeat.inputs:
  # 采集 ChatOps Bot 日志
  - type: container
    paths:
      - /var/lib/docker/containers/*chatops-bot*/*-json.log
    processors:
      - add_kubernetes_metadata:
          host: ${NODE_NAME}
          matchers:
            - logs_path:
                logs_path: "/var/lib/docker/containers/"

  # 采集网络设备 SSH 会话日志
  - type: log
    paths:
      - /app/logs/*.log
    multiline.pattern: '^\d{4}-\d{2}-\d{2}'
    multiline.negate: true
    multiline.match: after

# 输出到 Elasticsearch
output.elasticsearch:
  hosts: ["${ELASTICSEARCH_HOST:elasticsearch:9200}"]
  index: "chatops-logs-%{+yyyy.MM.dd}"

# 管道配置
setup.ilm.enabled: false
setup.template.name: "chatops"
setup.template.pattern: "chatops-logs-*"

# bot/log_collector.py - 容器内日志采集
"""
容器化环境下的日志采集与上报
"""
import logging
import json
import sys
from datetime import datetime
from logging.handlers import RotatingFileHandler


class ContainerLogHandler(logging.Handler):
    """
    容器日志处理器
    以 JSON 格式输出到 stdout，便于日志采集系统（Fluentd/Filebeat）收集
    """

    def __init__(self, service_name: str = "chatops-bot"):
        super().__init__()
        self.service_name = service_name
        self.writer = sys.stdout

    def emit(self, record: logging.LogRecord):
        try:
            log_entry = {
                'timestamp': datetime.utcnow().isoformat() + 'Z',
                'level': record.levelname,
                'service': self.service_name,
                'message': record.getMessage(),
                'module': record.module,
                'function': record.funcName,
                'line': record.lineno,
            }
            if record.exc_info and record.exc_info[0]:
                log_entry['exception'] = self.formatException(record.exc_info)

            # 输出 JSON 格式到 stdout
            self.writer.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
            self.writer.flush()
        except Exception:
            self.handleError(record)


def setup_logging(service_name: str = "chatops-bot",
                  level: int = logging.INFO):
    """配置容器化日志"""
    root_logger = logging.getLogger()
    root_logger.setLevel(level)

    # 移除默认处理器
    for handler in root_logger.handlers[:]:
        root_logger.removeHandler(handler)

    # 添加容器日志处理器（JSON 格式输出到 stdout）
    container_handler = ContainerLogHandler(service_name=service_name)
    container_handler.setFormatter(logging.Formatter('%(message)s'))
    root_logger.addHandler(container_handler)

    # 同时写入本地文件（用于容器内调试）
    file_handler = RotatingFileHandler(
        '/app/logs/app.log',
        maxBytes=50 * 1024 * 1024,  # 50MB
        backupCount=5,
        encoding='utf-8'
    )
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s [%(levelname)s] %(name)s:%(lineno)d - %(message)s'
    ))
    root_logger.addHandler(file_handler)


# K8s 日志采集 Pipeline
# Container stdout → Filebeat/Fluentd → Logstash → Elasticsearch → Kibana

5.4 监控告警集成代码

# bot/metrics_exporter.py - Prometheus 指标导出
"""
Prometheus 指标导出器
暴露网络运维平台的运行指标
"""
from prometheus_client import start_http_server, Counter, Histogram, Gauge, Info
import time


# 指标定义
COMMAND_EXEC_TOTAL = Counter(
    'chatops_command_exec_total',
    '命令执行总次数',
    ['command_type', 'status']  # 标签
)

COMMAND_EXEC_DURATION = Histogram(
    'chatops_command_exec_duration_seconds',
    '命令执行耗时（秒）',
    ['command_type'],
    buckets=(0.1, 0.5, 1, 2, 5, 10, 30, 60)
)

ACTIVE_DEVICES = Gauge(
    'chatops_active_devices',
    '活跃设备数量'
)

DEVICE_HEALTH_SCORE = Gauge(
    'chatops_device_health_score',
    '设备健康评分',
    ['device_name']
)

BOT_INFO = Info(
    'chatops_bot',
    'ChatOps 机器人信息'
)


class MetricsCollector:
    """指标采集器"""

    def __init__(self):
        BOT_INFO.info({
            'version': '1.2.0',
            'platform': 'chatops-network'
        })
        # 启动 Prometheus 指标服务（端口 9091）
        start_http_server(9091)

    def record_command(self, command_type: str, success: bool,
                      duration: float):
        """记录命令执行指标"""
        status = 'success' if success else 'failure'
        COMMAND_EXEC_TOTAL.labels(
            command_type=command_type, status=status
        ).inc()
        COMMAND_EXEC_DURATION.labels(
            command_type=command_type
        ).observe(duration)

    def update_device_count(self, count: int):
        """更新活跃设备数"""
        ACTIVE_DEVICES.set(count)

    def update_device_health(self, device_name: str, score: float):
        """更新设备健康评分"""
        DEVICE_HEALTH_SCORE.labels(
            device_name=device_name
        ).set(score)


# Grafana 告警规则示例
# monitoring/alerts.yml
"""
groups:
  - name: chatops_alerts
    rules:
      - alert: HighCommandFailureRate
        expr: |
          rate(chatops_command_exec_total{status="failure"}[5m])
          / rate(chatops_command_exec_total[5m]) > 0.1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "命令失败率超过 10%"
          description: "过去 5 分钟命令失败率 >10%"

      - alert: DeviceHealthCritical
        expr: chatops_device_health_score < 50
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "设备健康评分低于 50"
          description: "{{ $labels.device_name }} 健康评分: {{ $value }}"

      - alert: SlowCommandExecution
        expr: |
          histogram_quantile(0.95, rate(chatops_command_exec_duration_seconds_bucket[5m])) > 30
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "命令执行 P95 延迟超过 30 秒"
"""

六、总结

思考题

1. 通过聊天机器人执行网络操作时，如何确保操作的安全性（如权限控制、操作确认、防误操作）？
2. ChatOps 适合哪些运维场景？哪些操作不应通过 ChatOps 执行？

下篇预告：第328篇《大模型辅助网络故障诊断实践》——利用大语言模型（LLM）辅助网络故障根因分析与处置方案生成。