1. 项目背景与核心价值
每次凌晨三点被MongoDB集群告警吵醒时，我都会想：要是能把这些重复性的维护工作自动化该多好。经过三年在生产环境中的实践迭代，我总结出一套完整的MongoDB自动化运维方案，现在就把这些实战经验分享给你。
这套脚本体系主要解决DBA日常工作中的三大痛点：
- 重复性操作耗时（如定期压缩、索引重建）
- 人工操作风险高（特别是生产环境批量操作）
- 应急响应速度慢（故障时需快速获取诊断信息）
2. 核心脚本架构设计
2.1 技术选型决策
选择Python作为主要开发语言基于三个考量：
- PyMongo驱动成熟度（与MongoDB协议版本同步更新）
- 跨平台兼容性（Windows/Linux服务器通用）
- 丰富的运维生态（可集成Ansible/SaltStack）
# 典型脚本结构示例 class MongoMaintenance: def __init__(self, uri): self.client = MongoClient(uri, serverSelectionTimeoutMS=5000, socketTimeoutMS=30000) self.db = client.get_database('admin') def run_maintenance(self): try: self.check_connectivity() self.collection_compaction() self.index_rebuild() except Exception as e: self.alert_ops_team(str(e))2.2 模块化设计原则
将功能拆分为独立模块，通过配置文件动态加载：
/scripts ├── compaction.py # 集合压缩 ├── index_mgmt.py # 索引管理 ├── backup.py # 热备份 └── diagnostics.py # 健康检查重要提示：每个脚本必须实现统一的接口规范，包括--dry-run模式和日志分级输出
3. 关键维护场景实现
3.1 自动化压缩方案
集合碎片化超过30%时触发自动压缩：
def auto_compact(collections): for coll in collections: stats = db.command('collStats', coll['name']) fragmentation = 1 - (stats['size'] / stats['storageSize']) if fragmentation > 0.3: db.command('compact', coll['name'], force=True, paddingFactor=1.1) log(f"Compacted {coll['name']}, " f"reclaimed {fragmentation*100:.1f}% space")避坑经验：
- 避免业务高峰执行（通过cron设置时间窗口）
- 分片集群需单独处理每个shard
- 压缩期间会阻塞所有读写操作
3.2 智能索引维护
基于查询模式分析自动优化索引：
def optimize_indexes(): slow_queries = db.system.profile.find( {'millis': {'$gt': 100}}, {'ns':1, 'query':1}).distinct('ns') for ns in slow_queries: suggest_index(ns) # 使用$indexStats分析模式 rebuild_index(ns) # 在线重建索引典型优化效果对比：
| 操作类型 | 执行时间 | 锁占用时间 |
|---|---|---|
| 前台重建 | 42min | 42min |
| 后台重建 | 68min | 2.3s |
4. 生产环境实战技巧
4.1 安全防护机制
必须实现的三大安全措施：
- 操作审批流程（敏感操作需二次确认）
if danger_level > 3: send_approval_request(task_details) wait_for_approval(timeout=3600)- 自动熔断机制（当错误率>5%时停止任务）
error_rate = failed_ops / total_ops if error_rate > 0.05: alert("熔断触发！错误率已达" f"{error_rate*100}%") raise CircuitBreakerError- 操作回滚预案（所有变更操作记录undo脚本）
4.2 性能优化要点
批量操作时的关键参数：
bulk = db.collection.initialize_unordered_bulk_op() for doc in docs_to_update: bulk.find({'_id': doc['_id']}).update(doc) # 每1000条执行一次批量提交 bulk.execute({'w': 'majority', 'wtimeout': 5000, 'j': True})实测数据：批量提交比单条操作快17-23倍
5. 诊断工具集开发
5.1 实时健康检查
def health_check(): return { 'connections': db.serverStatus()['connections'], 'opcounters': db.serverStatus()['opcounters'], 'repl': db.admin.command('replSetGetStatus'), 'storage': db.stats(1024*1024) # MB单位 }5.2 性能瓶颈分析
使用$currentOp捕获慢操作：
db.adminCommand({ aggregate: 1, pipeline: [ {$currentOp: {allUsers: true}}, {$match: {secs_running: {$gt: 5}}}, {$project: {opid:1, ns:1, command:1}} ], cursor: {} })6. 部署与监控方案
6.1 分布式任务调度
通过消息队列实现跨集群任务分发：
[生产者] --(任务消息)--> RabbitMQ --(消费)--> [MongoDB集群A] `--(消费)--> [MongoDB集群B]6.2 Prometheus监控集成
关键指标采集配置：
- job_name: 'mongodb_exporter' static_configs: - targets: ['mongodb-exporter:9216'] metrics_path: '/metrics'核心监控指标清单：
- 连接数使用率
- 复制延迟秒数
- 操作队列长度
- 缓存命中率
这套脚本体系在我们生产环境每天自动处理300+维护任务，使DBA团队从重复劳动中解放出来。建议从最简单的健康检查脚本开始，逐步扩展功能模块。所有脚本已开源在GitHub（示例仓库见文末），你可以基于实际需求进行二次开发。