简介：这是一套面向IT运维与资产管理人员、系统集成开发者的CMDB资产配置管理系统源码包，适合用于学习配置管理数据库的落地实现，或作为企业内部资产集中管理、自动发现与关系映射功能的二次开发基础。压缩包共446个文件，约4.51MB，以Python源码与编译文件为主（98个py、117个pyc），辅以前端资源如80个js、42个map、21个css及svg、字体等，另含SQL建表脚本、配置文件与说明文档，覆盖后端逻辑、前端界面与数据库结构。系统围绕资产自动发现、配置项关系映射、数据同步集成、访问控制与审计等模块展开，并预留云计算与虚拟化环境下的资产管理扩展思路。目前已有65人学习下载，读者可从中获取完整的项目目录结构、可运行的代码骨架与数据库脚本，便于快速理解CMDB核心机制并搭建自己的资产管理原型。
1. 从一张 Excel 台账到 CMDB：资产配置管理系统到底在管什么
机房巡检时发现一台数据库服务器磁盘告警，运维翻出三个月前的 Excel 台账，上面写着「负责人：张三」，可张三上个月已经转岗。再查监控系统，这台机器的 IP 和台账里登记的完全对不上。这种场景几乎每个运维团队都遇到过——资产信息散落在 Excel、监控平台、工单系统和某位老员工的脑子里，一旦要查一台机器的完整信息，就得挨个系统翻。基于 CMDB 的资产配置管理系统，解决的就是这个问题：把服务器、网络设备、数据库实例、应用服务这些资源的属性、关系和变更历史，统一收进一个可查询、可追溯、可联动的配置管理数据库里。它适合运维工程师、SRE、IT 资产管理员，以及正在从「手工台账」往「自动化运维」过渡的中小团队。这篇文章不讲概念，直接讲怎么用 Python + Flask + MySQL 搭一套能跑起来的 CMDB 资产配置管理系统，包括数据模型怎么设计、API 怎么写、和监控系统怎么联动，以及我在实际部署中踩过的坑。
2. 数据模型设计：配置项、关系与变更记录怎么建表
2.1 为什么不能只用一张 assets 表
很多团队一开始图省事，建一张assets表，字段从 IP、主机名一路加到 CPU、内存、负责人、机房、机柜。用了半年就发现三个问题：第一，不同资产类型的属性差异太大，网络设备要记端口数、VLAN 划分，数据库实例要记版本、字符集，全塞一张表里大量字段为空；第二，资产之间的关系没法表达，比如「这台虚拟机跑在哪台宿主机上」「这个应用依赖哪个数据库实例」；第三，变更历史丢失，谁在什么时候改了 IP，查不到。
CMDB 的常见做法是拆成三张核心表：配置项表（CI）、关系表（Relation）、变更记录表（ChangeLog）。配置项表存资产的基本属性和类型，关系表存 CI 之间的依赖和归属，变更记录表存每次修改的快照。下面是我一般会用的建表语句。
-- 配置项表：所有资产的基础信息 CREATE TABLE ci ( id BIGINT PRIMARY KEY AUTO_INCREMENT, ci_type VARCHAR(32) NOT NULL COMMENT '类型：server/network/db/app', name VARCHAR(128) NOT NULL COMMENT '资产名称', ip VARCHAR(45) DEFAULT NULL COMMENT 'IPv4或IPv6地址', status TINYINT DEFAULT 1 COMMENT '1在线 0离线 2维护中', owner VARCHAR(64) DEFAULT NULL COMMENT '负责人', tags JSON DEFAULT NULL COMMENT '扩展属性，按类型存不同字段', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_type_name (ci_type, name), KEY idx_ip (ip) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 关系表：CI之间的依赖、归属、连接关系 CREATE TABLE ci_relation ( id BIGINT PRIMARY KEY AUTO_INCREMENT, source_id BIGINT NOT NULL COMMENT '源CI的id', target_id BIGINT NOT NULL COMMENT '目标CI的id', relation_type VARCHAR(32) NOT NULL COMMENT 'runs_on/depends_on/connects_to', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_rel (source_id, target_id, relation_type), KEY idx_target (target_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 变更记录表：每次修改留快照 CREATE TABLE ci_changelog ( id BIGINT PRIMARY KEY AUTO_INCREMENT, ci_id BIGINT NOT NULL, field_name VARCHAR(64) NOT NULL, old_value TEXT, new_value TEXT, operator VARCHAR(64) NOT NULL, changed_at DATETIME DEFAULT CURRENT_TIMESTAMP, KEY idx_ci (ci_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;ci表的tags字段用 JSON 类型，是为了解决不同资产类型属性差异的问题。服务器存{"cpu": 8, "mem_gb": 32, "os": "CentOS 7.9"}，数据库实例存{"version": "5.7.36", "charset": "utf8mb4"}，查询时用 MySQL 的JSON_EXTRACT函数就能按属性过滤。ci_relation表的relation_type我一般只保留三种：runs_on表示运行在什么之上，depends_on表示依赖什么，connects_to表示网络连接。关系类型不要设计太多，否则维护成本会失控。
2.2 用 Python 定义模型和自动记录变更
建完表之后，用 SQLAlchemy 定义模型，并在更新操作里自动写变更记录。下面这段代码是核心逻辑。
from sqlalchemy import Column, BigInteger, String, JSON, DateTime, func from sqlalchemy.ext.declarative import declarative_base from datetime import datetime Base = declarative_base() class CI(Base): __tablename__ = 'ci' id = Column(BigInteger, primary_key=True, autoincrement=True) ci_type = Column(String(32), nullable=False) name = Column(String(128), nullable=False) ip = Column(String(45)) status = Column(String(16), default='online') owner = Column(String(64)) tags = Column(JSON) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) def update_ci_with_changelog(session, ci_id, operator, **kwargs): """更新CI并自动记录每个字段的变更""" ci = session.query(CI).filter(CI.id == ci_id).first() if not ci: raise ValueError(f"CI {ci_id} not found") for field, new_val in kwargs.items(): old_val = getattr(ci, field, None) if str(old_val) != str(new_val): # 写变更记录 log = CIChangeLog( ci_id=ci_id, field_name=field, old_value=str(old_val), new_value=str(new_val), operator=operator ) session.add(log) setattr(ci, field, new_val) session.commit() return ci这段代码的关键点在于：变更记录是在字段级别做的，不是整条记录级别。这样查历史时可以直接问「谁改了这台机器的 IP」，而不是翻出一大坨 JSON 自己对比。operator参数必须从认证系统传入，不能由前端随便填，否则变更记录就失去了审计意义。tags字段因为是 JSON 类型，比较时用str()转换后再比，虽然粗糙但够用；如果对 JSON 内部字段的变更也要精确记录，需要单独展开处理，这个后面在避坑章节会讲。
3. 从零搭一套可查询的资产 API：Flask 路由与查询优化
3.1 最小可用的 Flask 接口
数据模型建好之后，下一步是让其他系统能查到资产信息。我一般用 Flask 起一个轻量 API 服务，核心接口就四个：按条件查资产列表、查单个资产详情、创建资产、更新资产。下面是最小可用的代码。
from flask import Flask, request, jsonify from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker app = Flask(__name__) engine = create_engine('mysql+pymysql://user:pass@127.0.0.1:3306/cmdb?charset=utf8mb4') Session = sessionmaker(bind=engine) @app.route('/api/ci', methods=['GET']) def list_ci(): """按类型、IP、负责人、标签过滤资产""" session = Session() ci_type = request.args.get('type') ip = request.args.get('ip') owner = request.args.get('owner') tag_key = request.args.get('tag_key') tag_val = request.args.get('tag_val') query = session.query(CI) if ci_type: query = query.filter(CI.ci_type == ci_type) if ip: query = query.filter(CI.ip == ip) if owner: query = query.filter(CI.owner == owner) if tag_key and tag_val: # MySQL JSON字段查询 query = query.filter(CI.tags[tag_key].as_string() == tag_val) results = query.limit(200).all() session.close() return jsonify([{ 'id': c.id, 'type': c.ci_type, 'name': c.name, 'ip': c.ip, 'status': c.status, 'owner': c.owner, 'tags': c.tags } for c in results]) @app.route('/api/ci/<int:ci_id>', methods=['GET']) def get_ci(ci_id): """查单个资产详情，附带关系""" session = Session() ci = session.query(CI).filter(CI.id == ci_id).first() if not ci: session.close() return jsonify({'error': 'not found'}), 404 relations = session.query(CIRelation).filter( (CIRelation.source_id == ci_id) | (CIRelation.target_id == ci_id) ).all() session.close() return jsonify({ 'id': ci.id, 'type': ci.ci_type, 'name': ci.name, 'ip': ci.ip, 'status': ci.status, 'owner': ci.owner, 'tags': ci.tags, 'relations': [{'source': r.source_id, 'target': r.target_id, 'type': r.relation_type} for r in relations] })list_ci接口里对tags字段的查询用的是 SQLAlchemy 的 JSON 索引语法，底层会翻译成 MySQL 的JSON_EXTRACT。这里有个性能注意点：JSON 字段上的查询没法走普通索引，如果资产量超过五千条，按标签过滤会明显变慢。我一般会在高频查询的标签上建虚拟列加索引，或者干脆把最常用的几个属性（比如 OS 类型、机房位置）从 JSON 里提出来做成独立字段。
3.2 查询性能优化：索引和分页
资产量上来之后，列表接口不加分页会直接拖垮数据库。我一般强制要求page和page_size参数，默认每页 50 条，最大 200 条。同时，ci表上除了主键和唯一键，至少还要在ci_type、owner、status上建索引，因为这三个字段是最高频的过滤条件。
ALTER TABLE ci ADD INDEX idx_type_status (ci_type, status); ALTER TABLE ci ADD INDEX idx_owner (owner);组合索引idx_type_status能同时覆盖「查所有在线服务器」这类查询。如果查询条件是「查某类型下某负责人的资产」，那idx_type_status加上owner的单列索引，MySQL 优化器一般会选组合索引先过滤类型和状态，再回表过滤负责人。更极致的做法是建(ci_type, owner, status)三列组合索引，但索引不是越多越好，每多一个索引就多一份写入开销，CMDB 的写入频率虽然不高，但批量导入时差距很明显。
分页查询用LIMIT offset, size在深分页时性能很差，比如LIMIT 10000, 50需要扫描前一万条。我一般用游标分页替代：前端传上一页最后一条的id，后端用WHERE id > last_id LIMIT 50。这样不管翻到第几页，查询速度都一样。
提示：如果团队里有人习惯直接连数据库查资产，建议把只读账号的权限限制在特定视图上，避免他们写出全表扫描的慢查询影响线上服务。
4. 和监控、工单系统联动：自动发现与数据同步
4.1 从监控系统自动同步资产
CMDB 最大的价值不是「能查」，而是「数据自动更新」。手工维护的台账活不过三个月，这是血泪经验。我一般会写一个定时任务，从监控系统（比如 Zabbix、Prometheus）的 API 拉取主机列表，和 CMDB 里的数据做对比，自动新增或标记离线。
import requests from datetime import datetime def sync_from_zabbix(api_url, token, session): """从Zabbix拉取主机列表，同步到CMDB""" headers = {'Content-Type': 'application/json-rpc'} payload = { 'jsonrpc': '2.0', 'method': 'host.get', 'params': { 'output': ['hostid', 'host', 'name', 'status'], 'selectInterfaces': ['ip', 'port'] }, 'auth': token, 'id': 1 } resp = requests.post(api_url, json=payload, headers=headers, timeout=10) hosts = resp.json().get('result', []) for h in hosts: ip = h['interfaces'][0]['ip'] if h['interfaces'] else None name = h['name'] # 按名称+IP查找已有CI ci = session.query(CI).filter(CI.name == name).first() if ci: # 已存在，更新IP和状态 if ci.ip != ip: update_ci_with_changelog(session, ci.id, 'sync_bot', ip=ip) ci.status = 'online' if h['status'] == '0' else 'offline' else: # 新增CI new_ci = CI(ci_type='server', name=name, ip=ip, status='online' if h['status'] == '0' else 'offline', owner='待分配', tags={'source': 'zabbix'}) session.add(new_ci) session.commit()这段同步逻辑的关键在于「以监控系统为准还是以 CMDB 为准」。我的做法是：IP 和在线状态以监控系统为准，负责人和业务标签以 CMDB 为准。同步时只更新监控系统能提供的字段，不覆盖人工维护的字段。sync_bot作为操作人写入变更记录，方便追溯哪些改动是自动同步产生的。
4.2 工单系统触发资产变更
另一个常见联动场景是：工单系统里申请了一台新服务器，审批通过后自动在 CMDB 里创建一条 CI 记录。这个流程一般用 webhook 实现——工单系统审批通过后回调 CMDB 的创建接口。
@app.route('/api/webhook/ticket_approved', methods=['POST']) def ticket_approved(): """工单审批通过后自动创建CI""" data = request.get_json() # 校验来源合法性，实际部署时用签名或内网IP白名单 if not verify_source(request): return jsonify({'error': 'forbidden'}), 403 session = Session() new_ci = CI( ci_type=data.get('asset_type', 'server'), name=data['hostname'], ip=data.get('ip'), status='maintenance', owner=data['applicant'], tags={'ticket_id': data['ticket_id'], 'source': 'ticket'} ) session.add(new_ci) session.commit() ci_id = new_ci.id session.close() return jsonify({'ci_id': ci_id, 'status': 'created'})这里有个容易翻车的地方：工单系统可能重复回调。如果网络抖动导致工单系统重试，CMDB 里就会出现同名资产。解决办法是在ci表的(ci_type, name)唯一键上做冲突处理，捕获IntegrityError后返回已有记录的 ID，而不是报错。
注意：webhook 接口必须做来源校验，至少限制内网 IP 段，否则任何人都能往 CMDB 里塞数据。我见过一个团队因为没做校验，被扫描器发现接口后灌了几万条垃圾数据，清理了一整天。
5. 避坑与排查：资产配置管理系统落地时最容易翻车的五件事
5.1 现象：同步任务跑完后，资产数量翻倍
原因：同步逻辑用的是「先查后插」，但查询条件只用了name，而不同机房的服务器可能重名。比如两个机房各有一台web-01，同步时第二台查不到就新建，结果 CMDB 里出现两条web-01。
解决：唯一键必须用(ci_type, name, ip)或者引入外部系统的主机 ID 作为唯一标识。如果监控系统有hostid，直接用它做external_id字段，同步时按external_id查，不要按名称查。
5.2 现象：JSON 字段里的属性改了，变更记录里看不到
原因：前面代码里update_ci_with_changelog对tags字段的比较是str(old_val) != str(new_val)，如果只改了 JSON 内部的一个 key，整个 JSON 字符串确实会变，变更记录会写。但如果用 SQLAlchemy 的MutableDict或者直接原地修改 JSON 对象，SQLAlchemy 可能检测不到变化，不会触发onupdate。
解决：更新tags时始终整体替换，不要原地修改。如果确实需要精确记录 JSON 内部字段的变更，在应用层展开对比，把每个 key 的变化单独写一条变更记录。
5.3 现象：按标签查询越来越慢，最后超时
原因：JSON 字段上的查询无法走索引，资产量到一万条以上时，每次按标签过滤都是全表扫描。
解决：把高频查询的标签提成独立字段。比如os_type、idc、env这三个标签，90% 的查询都围绕它们，直接在ci表加三列，建组合索引。低频标签继续留在 JSON 里，接受偶尔的慢查询。
5.4 现象：关系表里的数据越积越多，查一台机器的完整依赖要递归好几层
原因：ci_relation表只存了直接关系，查「这台应用依赖的所有数据库」需要递归查询。MySQL 8.0 之前不支持递归 CTE，应用层递归又容易写出 N+1 查询。
解决：如果 MySQL 版本是 8.0 以上，用WITH RECURSIVE写递归查询。如果是 5.7，在应用层做广度优先遍历，但一定要限制递归深度（比如最多 5 层），否则遇到环形依赖会死循环。更彻底的做法是单独建一张闭包表，把间接关系也存下来，查询时直接查闭包表，代价是关系变更时要维护闭包表的一致性。
5.5 现象：批量导入 Excel 时，部分数据导入失败但没有任何提示
原因：批量导入一般用session.add_all()然后一次性commit()，如果中间有一条数据违反唯一键约束，整个事务回滚，但代码里没有捕获异常，前端只看到「导入完成」，实际一条都没进去。
解决：批量导入时逐条commit()，或者用savepoint做部分回滚。每条失败的数据记录行号和失败原因，导入结束后返回给用户。我一般会在导入接口里返回一个{success: 120, failed: 3, errors: [{row: 45, reason: 'duplicate name'}]}的结构，让用户能直接定位问题行。
6. 进阶技巧：用 CMDB 数据驱动自动化巡检与容量分析
CMDB 建好之后，如果只是用来「查」，价值有限。真正让团队觉得「这东西值得投入」的，是把它和自动化脚本、容量分析打通。我一般会做两件事：自动生成巡检清单和容量趋势报表。
自动巡检的思路是：从 CMDB 查出所有status=online且ci_type=server的资产，按tags里的os字段分组，对 Linux 机器批量执行磁盘、内存、负载检查，对 Windows 机器走另一套脚本。下面是一个简化的巡检触发逻辑。
def generate_inspection_tasks(session): """从CMDB生成巡检任务清单""" servers = session.query(CI).filter( CI.ci_type == 'server', CI.status == 'online' ).all() tasks = [] for s in servers: os_type = (s.tags or {}).get('os', 'unknown') if 'CentOS' in os_type or 'Ubuntu' in os_type: script = 'inspect_linux.sh' elif 'Windows' in os_type: script = 'inspect_win.ps1' else: script = 'inspect_generic.sh' tasks.append({ 'ci_id': s.id, 'ip': s.ip, 'script': script, 'owner': s.owner }) return tasks这个清单可以直接推给 Ansible 或 SaltStack 执行。关键是tags里的os字段要准确，如果监控系统同步过来的数据里没有 OS 信息，可以在首次纳管时让负责人补录，或者通过 SSH 探测后回写。
容量分析则是定期统计各机房、各业务线的资产数量和资源规格，输出趋势。比如每月统计一次tags里cpu和mem_gb的总和，按owner分组，就能看出哪个业务线的资源增长最快。这个报表不需要复杂的代码，一条 SQL 就能出。
SELECT owner, COUNT(*) AS server_count, SUM(CAST(JSON_EXTRACT(tags, '$.cpu') AS UNSIGNED)) AS total_cpu, SUM(CAST(JSON_EXTRACT(tags, '$.mem_gb') AS UNSIGNED)) AS total_mem_gb FROM ci WHERE ci_type = 'server' AND status = 'online' GROUP BY owner ORDER BY total_cpu DESC;这条 SQL 在资产量五千条以内响应很快，超过之后建议把统计结果缓存到单独的报表表里，每天凌晨跑一次，前端查缓存。
最后说一个我自己的习惯：每次 CMDB 上线新功能，先拿一台测试机跑一周，确认变更记录、同步任务、查询接口都稳定了，再全量推。CMDB 是运维的地基，地基不稳，上面盖的自动化全是空中楼阁。希望帮到你。
本文还有配套的精品资源，点击获取
