第327篇：网络自动化运维平台架构设计

关键词

运维平台、平台架构、微服务、自动化引擎、统一门户、API 网关、任务调度、运维中台

一、为什么需要统一平台

1.1 脚本式自动化的局限

从脚本到平台的演进：

阶段 1：脚本自动化（单点突破） ┌──────────────────────────────────────────┐ │ backup.py ← 只做配置备份 │ │ patrol.py ← 只做巡检 │ │ deploy.py ← 只做部署 │ │ check.py ← 只做合规检查 │ │ │ │ 问题： │ │ ┌─ 每个脚本独立运行 │ │ ├─ 没有统一的凭证管理 │ │ ├─ 没有调度和编排 │ │ ├─ 没有用户权限控制 │ │ ├─ 没有审计日志 │ │ └─ 工程师还是需要记住所有脚本名 │ └──────────────────────────────────────────┘

阶段 2：平台化（能力集结） | 网络自动化运维平台 ┌────────────────────────────────────┐ | 统一门户（Web UI + API） ┌─ 备份 → 无感执行 ├─ 巡检 → 一键触发 ├─ 部署 → 审批后自动执行 └─ 合规 → 定时自动检查 平台提供： ┌─ 统一凭证管理（加密存储） ├─ 设备目录（CMDB 集成） ├─ 任务调度（定时/触发） ├─ 权限控制（RBAC） ├─ 审计日志 └─ 标准化输出（不关心厂商差异） | | | --- | --- | --- |

1.2 平台架构目标

平台架构核心目标：

  1. 统一入口
  ┌─ Web UI 操作（降低使用门槛）
  ├─ REST API 集成（供其他系统调用）
  └─ ChatOps 命令（IM 中执行）

  2. 能力沉淀
  ┌─ 将可复用的运维能力封装为服务
  ├─ 新需求通过编排现有服务实现
  └─ 避免重复造轮子

  3. 安全可控
  ┌─ 用户权限最小化（RBAC）
  ├─ 敏感信息加密（设备凭证）
  ├─ 操作可追溯（完整审计日志）
  └─ 变更审批流程（双人复核）

  4. 可扩展
  ┌─ 插件式架构（新能力即插即用）
  ├─ 标准化接口（降低集成成本）
  └─ 水平扩展（支撑大规模设备）

二、平台架构设计

2.1 整体架构

网络自动化运维平台架构：

2.2 核心数据模型

#!/usr/bin/env python3
# platform_models.py — 平台核心数据模型

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from enum import Enum
from datetime import datetime
import json


# =============================================
# 设备管理
# =============================================

@dataclass
class Device:
    """设备信息"""
    id: str                    # 设备 ID（UUID）
    hostname: str              # 主机名
    management_ip: str         # 管理 IP
    device_type: str           # 厂商+型号（huawei_vrp）
    vendor: str                # 厂商
    model: str                 # 型号
    os_version: str            # 系统版本
    role: str                  # 角色（core/access/border）
    location: str = ""         # 物理位置
    status: str = "active"     # 状态
    tags: List[str] = field(default_factory=list)
    created_at: str = field(
        default_factory=lambda: datetime.now().isoformat()
    )


@dataclass
class DeviceCredential:
    """设备凭证（加密存储）"""
    device_id: str
    username: str
    password: str              # 加密存储
    enable_password: str = ""  # 加密存储
    ssh_key: str = ""          # 加密存储
    updated_at: str = field(
        default_factory=lambda: datetime.now().isoformat()
    )


# =============================================
# 任务系统
# =============================================

class TaskStatus(Enum):
    PENDING   = "pending"
    RUNNING   = "running"
    SUCCESS   = "success"
    FAILED    = "failed"
    CANCELLED = "cancelled"


class TaskType(Enum):
    BACKUP      = "backup"
    PATROL      = "patrol"
    DEPLOY      = "deploy"
    COMPLIANCE  = "compliance"
    PING        = "ping"
    CUSTOM      = "custom"


@dataclass
class Task:
    """任务定义"""
    id: str
    type: TaskType
    name: str
    devices: List[str]         # 目标设备 ID 列表
    params: Dict               # 任务参数
    schedule: str = ""         # cron 表达式（定时任务）
    status: TaskStatus = TaskStatus.PENDING
    created_by: str = ""
    created_at: str = field(
        default_factory=lambda: datetime.now().isoformat()
    )
    started_at: Optional[str] = None
    completed_at: Optional[str] = None
    result: Dict = field(default_factory=dict)


@dataclass
class TaskResult:
    """任务执行结果"""
    task_id: str
    device_id: str
    status: TaskStatus
    output: Dict = field(default_factory=dict)
    error: str = ""
    started_at: Optional[str] = None
    completed_at: Optional[str] = None
    duration_ms: int = 0


# =============================================
# 审计日志
# =============================================

@dataclass
class AuditLog:
    """审计日志"""
    id: str
    user: str                  # 操作人
    action: str                # 操作类型
    resource_type: str         # 资源类型
    resource_id: str           # 资源 ID
    detail: str                # 操作详情
    ip_address: str            # 来源 IP
    status: str                # 成功/失败
    timestamp: str = field(
        default_factory=lambda: datetime.now().isoformat()
    )
    extra: Dict = field(default_factory=dict)

三、核心服务实现

3.1 任务调度服务

#!/usr/bin/env python3
# scheduler_service.py — 任务调度服务

from typing import List, Dict, Optional
from datetime import datetime
import asyncio
import json
import logging

logger = logging.getLogger(__name__)


class TaskScheduler:
    """任务调度器"""

    def __init__(self, executor=None):
        self.executor = executor
        self._tasks: Dict[str, Task] = {}
        self._running: Dict[str, bool] = {}

    def create_task(
        self,
        task_type: TaskType,
        name: str,
        devices: List[str],
        params: Dict,
        schedule: str = "",
        created_by: str = "",
    ) -> Task:
        """创建任务"""
        import uuid
        task = Task(
            id=str(uuid.uuid4()),
            type=task_type,
            name=name,
            devices=devices,
            params=params,
            schedule=schedule,
            created_by=created_by,
        )
        self._tasks[task.id] = task
        logger.info(f"创建任务: {task.id} - {name}")
        return task

    def execute_task(self, task_id: str) -> Dict:
        """执行任务"""
        task = self._tasks.get(task_id)
        if not task:
            return {"error": f"任务 {task_id} 不存在"}

        task.status = TaskStatus.RUNNING
        task.started_at = datetime.now().isoformat()
        results = {}

        for device_id in task.devices:
            try:
                device_result = self.executor.execute(
                    device_id, task.type, task.params
                )
                results[device_id] = device_result
                logger.info(
                    f"任务 {task_id} 设备 {device_id} 执行完成"
                )
            except Exception as e:
                results[device_id] = {"error": str(e)}
                logger.error(
                    f"任务 {task_id} 设备 {device_id} 失败: {e}"
                )

        task.result = results
        task.completed_at = datetime.now().isoformat()

        # 判断整体状态
        all_success = all(
            "error" not in r for r in results.values()
        )
        task.status = (
            TaskStatus.SUCCESS if all_success
            else TaskStatus.FAILED
        )

        return results

    def get_task(self, task_id: str) -> Optional[Task]:
        """获取任务信息"""
        return self._tasks.get(task_id)

    def list_tasks(
        self, status: Optional[TaskStatus] = None
    ) -> List[Task]:
        """列出任务"""
        if status:
            return [
                t for t in self._tasks.values()
                if t.status == status
            ]
        return list(self._tasks.values())

    def cancel_task(self, task_id: str) -> bool:
        """取消任务"""
        task = self._tasks.get(task_id)
        if task and task.status in (
            TaskStatus.PENDING, TaskStatus.RUNNING
        ):
            task.status = TaskStatus.CANCELLED
            return True
        return False

3.2 执行引擎服务

class ExecutionEngine:
    """执行引擎"""

    def __init__(self):
        self._adapters = {}

    def register_adapter(
        self, device_type: str, adapter
    ):
        """注册设备适配器"""
        self._adapters[device_type] = adapter

    def execute(
        self, device_id: str,
        task_type: TaskType,
        params: Dict,
    ) -> Dict:
        """在设备上执行任务"""
        # 根据 device_id 获取设备信息和适配器
        device = self._get_device(device_id)
        if not device:
            return {"error": f"设备 {device_id} 不存在"}

        adapter = self._adapters.get(device.device_type)
        if not adapter:
            return {
                "error": f"不支持的设备类型: {device.device_type}"
            }

        # 根据任务类型分发
        handlers = {
            TaskType.BACKUP: self._handle_backup,
            TaskType.PATROL: self._handle_patrol,
            TaskType.DEPLOY: self._handle_deploy,
            TaskType.PING: self._handle_ping,
        }

        handler = handlers.get(task_type)
        if not handler:
            return {"error": f"不支持的任务类型: {task_type}"}

        return handler(adapter, params)

    def _handle_backup(self, adapter, params) -> Dict:
        config = adapter.get_config()
        return {
            "config_length": len(config),
            "timestamp": datetime.now().isoformat(),
            "config": config,
        }

    def _handle_patrol(self, adapter, params) -> Dict:
        facts = adapter.get_facts()
        interfaces = adapter.get_interfaces()
        bgp = adapter.get_bgp_peers()
        return {
            "facts": facts,
            "interfaces": interfaces,
            "bgp": bgp,
            "timestamp": datetime.now().isoformat(),
        }

    def _handle_deploy(self, adapter, params) -> Dict:
        commands = params.get("commands", [])
        results = adapter.send_commands(commands)
        adapter.save_config()
        return {
            "commands_executed": len(commands),
            "results": results,
        }

    def _handle_ping(self, adapter, params) -> Dict:
        target = params.get("target", "")
        count = params.get("count", 3)
        return adapter.ping(target, count)

    def _get_device(self, device_id: str):
        """从数据库获取设备信息"""
        # TODO: 接入 CMDB 或本地数据库
        pass

3.3 Web API 服务

#!/usr/bin/env python3
# api_service.py — REST API 服务

from flask import Flask, request, jsonify
from flask_cors import CORS
import jwt
import hashlib
import hmac
from functools import wraps

app = Flask(__name__)
CORS(app)


# =============================================
# 认证与权限
# =============================================

def token_required(f):
    """JWT 认证装饰器"""
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get("Authorization", "")
        try:
            payload = jwt.decode(
                token.split()[-1],
                app.config["SECRET_KEY"],
                algorithms=["HS256"],
            )
            request.user = payload
        except Exception:
            return jsonify({"error": "认证失败"}), 401
        return f(*args, **kwargs)
    return decorated


def role_required(roles: List[str]):
    """角色权限装饰器"""
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            user_role = request.user.get("role", "")
            if user_role not in roles:
                return jsonify({"error": "权限不足"}), 403
            return f(*args, **kwargs)
        return decorated
    return decorator


# =============================================
# API 路由
# =============================================

@app.route("/api/v1/devices", methods=["GET"])
@token_required
def list_devices():
    """获取设备列表"""
    # TODO: 从数据库查询
    return jsonify({"devices": []})


@app.route("/api/v1/devices/<device_id>", methods=["GET"])
@token_required
def get_device(device_id):
    """获取设备详情"""
    # TODO: 从数据库查询
    return jsonify({"device": {"id": device_id}})


@app.route("/api/v1/tasks", methods=["GET"])
@token_required
def list_tasks():
    """获取任务列表"""
    status = request.args.get("status")
    tasks = scheduler.list_tasks(
        TaskStatus(status) if status else None
    )
    return jsonify({
        "tasks": [
            {
                "id": t.id,
                "name": t.name,
                "type": t.type.value,
                "status": t.status.value,
                "created_at": t.created_at,
            }
            for t in tasks
        ]
    })


@app.route("/api/v1/tasks", methods=["POST"])
@token_required
def create_task():
    """创建任务"""
    data = request.get_json()
    task = scheduler.create_task(
        task_type=TaskType(data["type"]),
        name=data["name"],
        devices=data["devices"],
        params=data.get("params", {}),
        schedule=data.get("schedule", ""),
        created_by=request.user.get("username", ""),
    )
    # 立即执行
    if data.get("execute_now", True):
        scheduler.execute_task(task.id)
    return jsonify({"task_id": task.id}), 201


@app.route("/api/v1/tasks/<task_id>", methods=["GET"])
@token_required
def get_task(task_id):
    """获取任务详情"""
    task = scheduler.get_task(task_id)
    if not task:
        return jsonify({"error": "任务不存在"}), 404
    return jsonify({
        "id": task.id,
        "name": task.name,
        "type": task.type.value,
        "status": task.status.value,
        "devices": task.devices,
        "result": task.result,
        "created_at": task.created_at,
    })


# =============================================
# 启动
# =============================================

if __name__ == "__main__":
    scheduler = TaskScheduler()
    app.config["SECRET_KEY"] = "your-secret-key"
    app.run(host="0.0.0.0", port=5000, debug=True)

四、平台关键能力

4.1 凭证安全管理

设备凭证安全管理：

  存储策略：
  ┌──────────────────────────────────────────┐
  │  加密存储（AES-256-GCM）                  │
  │  ┌─ 凭证写入数据库前加密                 │
  │  ├─ 读取时解密                           │
  │  ├─ 密钥定期轮换                         │
  │  └─ 加密密钥与数据分离存储               │
  │                                           │
  │  访问控制（最小权限）                     │
  │  ┌─ 管理员：管理所有凭证                 │
  │  ├─ 操作员：使用凭证（不能查看明文）     │
  │  ├─ 审计员：查看访问日志（不能使用）     │
  │  └─ 只读用户：无权访问凭证               │
  │                                           │
  │  审计追踪                                │
  │  ┌─ 谁在何时使用了哪个凭证               │
  │  ├─ 谁解密查看了凭证明文                 │
  │  └─ 异常访问告警                         │
  │                                           │
  │  自动轮换                                │
  │  ┌─ 定期自动修改设备密码                 │
  │  ├─ 更新后同步到凭证库                   │
  │  └─ 变更审批流程                         │
  └──────────────────────────────────────────┘

4.2 审计与合规

class AuditService:
    """审计服务"""

    def __init__(self):
        self.logs: List[AuditLog] = []

    def log(
        self,
        user: str,
        action: str,
        resource_type: str,
        resource_id: str,
        detail: str,
        status: str = "success",
        ip: str = "",
    ):
        """记录审计日志"""
        import uuid
        log = AuditLog(
            id=str(uuid.uuid4()),
            user=user,
            action=action,
            resource_type=resource_type,
            resource_id=resource_id,
            detail=detail,
            status=status,
            ip_address=ip,
        )
        self.logs.append(log)
        # 异步写入数据库和 ElasticSearch

    def query(
        self,
        user: str = "",
        action: str = "",
        start_time: str = "",
        end_time: str = "",
        limit: int = 100,
    ) -> List[AuditLog]:
        """查询审计日志"""
        results = self.logs

        if user:
            results = [l for l in results if l.user == user]
        if action:
            results = [l for l in results if l.action == action]
        if start_time:
            results = [
                l for l in results if l.timestamp >= start_time
            ]
        if end_time:
            results = [
                l for l in results if l.timestamp <= end_time
            ]

        return results[:limit]

    def generate_audit_report(
        self, start_date: str, end_date: str
    ) -> Dict:
        """生成审计报告"""
        logs = self.query(
            start_time=start_date, end_time=end_date
        )

        return {
            "period": f"{start_date} ~ {end_date}",
            "total_operations": len(logs),
            "by_user": self._group_by("user", logs),
            "by_action": self._group_by("action", logs),
            "failed_operations": len(
                [l for l in logs if l.status == "failed"]
            ),
        }

    def _group_by(
        self, key: str, logs: List[AuditLog]
    ) -> Dict:
        result = {}
        for log in logs:
            val = getattr(log, key, "unknown")
            result[val] = result.get(val, 0) + 1
        return result

五、部署与运维

5.1 容器化部署

# docker-compose.yml — 平台容器化部署

version: '3.8'

services:
  api-gateway:
    image: nginx:alpine
    ports:
      - "8080:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - web-ui
      - api-service

  web-ui:
    build: ./frontend
    ports:
      - "3000:3000"
    environment:
      - API_URL=http://api-service:5000

  api-service:
    build: ./backend
    ports:
      - "5000:5000"
    environment:
      - DB_HOST=mysql
      - REDIS_HOST=redis
      - RABBITMQ_HOST=rabbitmq
    depends_on:
      - mysql
      - redis
      - rabbitmq

  task-worker:
    build: ./backend
    command: python worker.py
    environment:
      - DB_HOST=mysql
      - REDIS_HOST=redis
    depends_on:
      - mysql
      - redis
      - rabbitmq

  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: root123
      MYSQL_DATABASE: network_ops
    volumes:
      - mysql_data:/var/lib/mysql

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"

  influxdb:
    image: influxdb:2.0
    volumes:
      - influxdb_data:/var/lib/influxdb2

  elasticsearch:
    image: elasticsearch:8.10
    environment:
      - discovery.type=single-node
    volumes:
      - es_data:/usr/share/elasticsearch/data

volumes:
  mysql_data:
  influxdb_data:
  es_data:

5.2 平台运维指标

平台自身运维指标：

  1. 可用性
  ┌─ 平台服务可用率（目标：99.9%）
  ├─ API 响应时间（P99 < 2s）
  └─ 任务执行成功率（目标：> 99%）

  2. 容量
  ┌─ 支持的设备数量
  ├─ 并发任务数
  ├─ 日处理任务量
  └─ 存储使用量

  3. 安全性
  ┌─ 认证失败次数
  ├─ 权限越界尝试
  ├─ 凭证访问审计
  └─ API 异常请求

  4. 用户满意度
  ┌─ 任务完成时间
  ├─ 用户自助完成率
  ├─ 用户反馈评分
  └─ 培训通过率

六、最佳实践总结

自动化运维平台建设建议：

  1. 先有脚本，再有平台
  ┌─ 不要一开始就追求大而全的平台
  ├─ 先把高频、重复的操作脚本化
  ├─ 验证脚本的可靠性和效率
  └─ 再把优秀脚本迁移到平台

  2. 平台要"好用"而不是"强大"
  ┌─ UI 要简单直观（面向运维工程师）
  ├─ 常用操作一键完成
  ├─ 减少点击次数
  └─ 错误提示清晰明了

  3. 开放生态
  ┌─ 提供标准 API 接口
  ├─ 支持自定义插件
  ├─ 与其他系统集成（CMDB/工单/监控）
  └─ 数据开放可导出

  4. 渐进式建设
  ┌─ 第一个版本只做配置备份 + 巡检
  ├─ 第二个版本加入变更部署
  ├─ 第三个版本加入合规检查
  └─ 持续迭代，从不完美开始

  5. 运营度量
  ┌─ 跟踪平台使用率
  ├─ 跟踪自动化覆盖率
  ├─ 收集用户反馈
  └─ 展示自动化价值（节省人天）

下篇预告：第328篇 — 零接触部署与自动上线，讲解零接触部署（ZTP）的实现原理和企业落地方法。

下篇预告：第328篇 — 零接触部署与自动上线，讲解零接触部署（ZTP）的实现原理和企业落地方法。