基于Django自研云资源巡检工具：补足云监控盲区的实践

简介在云原生与基础设施规模扩张的背景下资源巡检逐渐成为运维体系中的关键环节。传统云监控专注于实时资源指标却难以覆盖配置合规、依赖健康与容量趋势等隐性风险。Django作为成熟的后端框架凭借ORM、Admin后台与完善的定时任务生态为企业自建巡检工具提供了高效底座。通过对接阿里云OpenAPI结合ECS、SLB等资产模型与注册式巡检引擎可实现安全组规则、证书到期、后端健康状态等自动检查同时利用APScheduler完成周期调度并联动钉钉、邮件实现告警触达。这套方案填补了云监控覆盖不到的盲区形成“实时监控周期体检”的互补机制。本文面向运维与后端开发者介绍从模型设计、API封装到告警降噪的完整路径帮助中小企业以较低成本建立属于自己的巡检能力。1. 为什么中小企业需要一套自己的巡检工具1.1 云监控覆盖不到的检查项先说下背景。我这边负责的团队大概有三十多台阿里云ECS和七八个SLB实例规模不大不小正好卡在人工能管和必须自动化的尴尬区间里。之前一直用阿里云自带的云监控配了几条CPU、内存、磁盘的告警规则以为就高枕无忧了。直到有一次线上事故让我改了这个想法。那次的起因很简单一台跑着Nginx的ECS磁盘满了日志把根分区塞到100%Nginx直接拒绝写入接口全部503。云监控确实是告警了但问题在于——阿里云云监控的默认告警阈值是80%的磁盘使用率而我们根本没收到通知因为告警联系人离职了钉钉群没有同步邮件进了垃圾箱。等到用户先发现故障来投诉我们才知道出了问题。后来我仔细盘点了一下发现云监控这类基础设施监控工具存在几个明显的盲区它监控的是资源是否健康不关心配置是否合规。比如你的安全组把22端口对0.0.0.0/0开放了云监控不会觉得有什么问题。它不感知业务层面的巡检项比如SLB后端某台ECS被摘除了、证书还有三天到期、某个Bucket的权限变成了公共读这些都是看不见的。告警触达链路太长配置复杂如果账号多、项目多很难统一管理。那你说直接用阿里云的配置审计我也想用但它按规则数量和资源数量收费对中小企业小几万的年费预算来说性价比真的很一般。而且配置审计主要是事后合规定位跟日常巡检的思路还不完全一样。总之我需要的是这样一套东西每天早上定时把ECS和SLB翻一遍检查磁盘、内存、安全组、证书、后端健康状态等关键项发现问题推到钉钉群并留一份历史记录方便追溯。既然市面上找不到完全贴合的现成工具那就自己写一个。1.2 用Django而不是Go或Shell脚本的原因可能有人会说巡检这种周期性任务写个Python脚本配合crontab不就完事了为什么非要上一个Django项目我最初也确实是这么干的写了几个独立的Python脚本crontab里排了一堆任务。但跑了两个月就遇到了问题脚本之间互相不通信资产清单是硬编码在Python列表里的上线一台新机器得改代码巡检结果只输出成日志文件想看历史趋势得手动去翻新增一个巡检项要在多个脚本里同步改。说白了没有一个统一的主心骨去管理资产、任务、结果这三样东西。Django在这个场景下的优势非常明显ORM和Admin后台是天然的资产管理系统。我在Admin里维护ECS、SLB清单写个Model就能自动获得增删改查界面不用单独开发管理页面。定时任务生态成熟。django-apscheduler、Celery Beat都能很好地和ORM融合任务执行结果可以落库天然具备可追溯性。代码组织清晰。每个巡检逻辑可以抽象成一个独立的类或函数注册到统一的巡检引擎里新增巡检项只需要写一个文件。当然用Go写也挺好性能更强、部署更简单。但考虑到团队同学都是Python背景Django的快速迭代优势更实际。工具是拿来用的不是拿来炫技的团队最熟悉什么就用什么这点很重要。1.3 自研巡检工具与云监控的互补关系这里需要明确一下边界自研巡检工具不是替代云监控的它们是两层东西。云监控负责实时性——资源指标每15秒采集一次CPU飙到99%能立刻告警。自研巡检工具负责的是周期性体检——每天定时过一遍配置项和状态项发现的是慢性病和隐性风险。打个比方云监控是医院的ICU监护仪盯着各项生命体征自研巡检工具是每年的体检套餐查的是血脂血压、潜在隐患。ICU的设备再全也不能帮你查有没有蛀牙。所以我在设计这套工具时特意没有去抓实时监控数据那些事情让云监控做更好。我只关注四类问题配置是否合规安全组、白名单、密钥对容量风险磁盘使用率、内存是否有泄漏趋势依赖是否健康SLB后端、证书有效期、域名解析状态资源是否浪费闲置实例、未绑定的公网IP按这个思路去规划功能就不会和云监控抢活儿干恰好补上它看不见的部分。2. Django项目骨架与核心数据模型设计2.1 项目结构和App划分这套工具我给它起了个内部代号叫Inspector。项目创建用的是django-admin startproject然后划分了四个App各管一摊inspector/ # 项目配置目录 settings.py urls.py celery_app.py # Celery实例初期用的Celery后来换APScheduler apps/ assets/ # 资产清单ECS、SLB实例的增删改查 inspect/ # 巡检引擎巡检项定义、执行、结果存储 alerts/ # 告警通知钉钉、邮件、飞书 accounts/ # 登录认证和操作审计 reports/ # 巡检报告和趋势分析 manage.py requirements.txt这个划分是我跑过一版单体脚本之后反思得到的。核心逻辑是资产是物质的巡检是动作告警是结果。如果不把资产和巡检拆开后面一旦想增加新的资产类型比如RDS、OSS、Redis就会牵一发动全身。2.2 核心模型资产表与巡检任务表Django的ORM建模型是强项这套工具的数据模型也成了整个系统的骨架。下面几个模型是我反复调整后定下来的给你参考。ECS资产表class EcsInstance(models.Model): instance_id models.CharField(max_length64, uniqueTrue, verbose_name实例ID) instance_name models.CharField(max_length128, verbose_name实例名称) region_id models.CharField(max_length32, verbose_name地域) status models.CharField(max_length16, verbose_name状态) instance_type models.CharField(max_length64, verbose_name规格) public_ip models.GenericIPAddressField(nullTrue, blankTrue, verbose_name公网IP) private_ip models.GenericIPAddressField(nullTrue, blankTrue, verbose_name内网IP) os_name models.CharField(max_length128, blankTrue, verbose_name操作系统) expired_at models.DateTimeField(nullTrue, blankTrue, verbose_name到期时间) created_at models.DateTimeField(auto_now_addTrue) updated_at models.DateTimeField(auto_nowTrue) class Meta: db_table asset_ecs ordering [-created_at]SLB资产表class SlbInstance(models.Model): load_balancer_id models.CharField(max_length64, uniqueTrue, verbose_nameSLB实例ID) load_balancer_name models.CharField(max_length128, verbose_nameSLB名称) address models.GenericIPAddressField(verbose_nameSLB地址) address_type models.CharField(max_length16, verbose_name地址类型) region_id models.CharField(max_length32, verbose_name地域) status models.CharField(max_length16, verbose_name状态) created_at models.DateTimeField(auto_now_addTrue) updated_at models.DateTimeField(auto_nowTrue) class Meta: db_table asset_slb巡检任务表class InspectTask(models.Model): name models.CharField(max_length128, verbose_name任务名称) task_type models.CharField(max_length32, choices[ (ecs, ECS巡检), (slb, SLB巡检), (oss, OSS巡检), ], verbose_name巡检类型) cron_expression models.CharField(max_length64, default0 8 * * *, verbose_name定时表达式) enabled models.BooleanField(defaultTrue, verbose_name是否启用) last_run_at models.DateTimeField(nullTrue, blankTrue, verbose_name上次执行时间) created_at models.DateTimeField(auto_now_addTrue) class Meta: db_table inspect_task巡检结果表class InspectResult(models.Model): task models.ForeignKey(InspectTask, on_deletemodels.CASCADE, verbose_name所属任务) instance_id models.CharField(max_length64, verbose_name实例ID) check_item models.CharField(max_length128, verbose_name巡检项名称) result models.CharField(max_length16, choices[ (pass, 通过), (warn, 警告), (fail, 失败), ], verbose_name巡检结果) detail models.TextField(blankTrue, verbose_name详情) raw_data models.JSONField(defaultdict, verbose_name原始数据) checked_at models.DateTimeField(auto_now_addTrue, verbose_name巡检时间) class Meta: db_table inspect_result indexes [ models.Index(fields[instance_id, check_item, checked_at]), ]这里有两个设计细节值得展开说说。第一个是raw_data字段。我把每次巡检的原始返回结果用JSON格式原样存下来而不是只存一个人眼可读的结论。这样做的价值在于当发现某个巡检项被判为fail你不用再调用一次阿里云OpenAPI去查原始数据直接看这个JSON就能定位问题。排查时间从小时级降到了分钟级。第二个是索引设计。巡检结果表会随着时间增长得很快我刚开始没在意这个结果跑了几个月之后查询InspectorAdmin页面按实例查历史结果这种非常常见的操作响应时间从几百毫秒涨到了十秒以上。加了复合索引之后同样的查询不到100毫秒就完成了。你在自己的项目里一定要注意巡检结果这种写入频繁、按查询条件筛选的表复合索引是必需品不是可选项。2.3 巡检项注册机制巡检引擎是整个系统的中枢它负责执行巡检任务、收集结果、触发告警。我用了一个注册-发现模式来管理巡检项让新增一个巡检项变得非常简单。每个巡检项是一个独立的Python类继承自一个基类然后实现check方法class BaseChecker: # 巡检项标识必须唯一 code # 巡检项名称用于展示 name # 适用的资源类型 resource_type # 告警级别info/warning/critical severity warning def __init__(self, client): self.client client def check(self, resource): 核心巡检逻辑接收一个资产对象返回一个InspectResult对象 raise NotImplementedError然后通过元类自动把所有子类注册到一个全局注册表里class CheckerRegistry: _checkers {} classmethod def register(cls, checker_class): cls._checkers[checker_class.code] checker_class return checker_class classmethod def get_all(cls, resource_typeNone): for code, checker_class in cls._checkers.items(): if resource_type and checker_class.resource_type ! resource_type: continue yield checker_class() class BaseChecker(metaclassCheckerMeta): ...实际使用的时候巡检引擎只需要遍历当前任务类型的注册列表逐个调用check方法把返回值写入InspectResult表即可。后面再想新增巡检项比如检查ECS的自动快照策略是否启用只需要写一个新类继承BaseChecker实现check方法然后在类上标注一个装饰器就能被自动发现并执行。完全不需要改动现有的引擎代码。这套设计早期花了一点时间但收益是长期的。后面我们增加了OSS Bucket权限检查、RDS白名单检查、Redis密码检查都是二十分钟就能上一个新巡检项。3. 阿里云OpenAPI对接与巡检执行的核心实现3.1 凭据管理与RAM子账号对接阿里云OpenAPI第一步要解决的是API鉴权。这里有一个绝对不能踩的坑不要把主账号的AccessKey写在代码或配置文件里。一旦泄露整个账号的云资源都可能被删光出事就是大事故。我的做法是创建一个只读权限的RAM子账号把权限控制在最小范围{ Statement: [ { Effect: Allow, Action: [ ecs:DescribeInstances, ecs:DescribeDisks, ecs:DescribeSecurityGroups, ecs:DescribeSecurityGroupAttribute, slb:DescribeLoadBalancers, slb:DescribeHealthStatus, slb:DescribeServerCertificates, slb:DescribeListenerAccessControlAttribute, rds:DescribeDBInstances, oss:GetBucketAcl ], Resource: * } ], Version: 1 }注意Action列表只列出了查询类API没有Create、Delete、Modify这些写操作。巡检工具只需要看永远不需要改。AccessKey的存储我用的是环境变量在systemd服务文件里注入配置文件里不出现任何真实凭据。如果你有更严格的安全要求可以接KMS或者RAM角色授权。对于中小企业环境变量RAM子账号已经是很稳的方案了。3.2 SDK封装与ECS巡检阿里云Python SDK的调用方式大同小异核心就三步初始化Client、构造Request、调用接口。我封装了一个统一的Client工厂from aliyunsdkcore.client import AcsClient from aliyunsdkcore.auth.credentials import AccessKeyCredential def get_acs_client(region_id): credentials AccessKeyCredential( os.environ[ALIYUN_AK_ID], os.environ[ALIYUN_AK_SECRET] ) return AcsClient( region_idregion_id, credentialcredentials )ECS巡检最核心的几个调用拉取ECS实例列表from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest def fetch_ecs_instances(client, page_size100): request DescribeInstancesRequest.DescribeInstancesRequest() request.set_PageSize(page_size) request.set_PageNumber(1) response client.do_action_with_exception(request) data json.loads(response) instances data.get(Instances, {}).get(Instance, []) return instances这里有个细节DescribeInstances返回的实例列表是分页的。如果你用默认的PageSize最多返回10条不处理翻页逻辑的话会把大量实例漏掉巡检结果看起来一切正常实际上很多机器根本没被检查到。我建议PageSize设成100然后判断返回的数量是否等于PageSize等于就继续翻页直到取完为止。获取安全组规则from aliyunsdkecs.request.v20140526.DescribeSecurityGroupAttributeRequest import DescribeSecurityGroupAttributeRequest def fetch_security_group_rules(client, security_group_id): request DescribeSecurityGroupAttributeRequest.DescribeSecurityGroupAttributeRequest() request.set_SecurityGroupId(security_group_id) response client.do_action_with_exception(request) data json.loads(response) return data.get(Permissions, {}).get(Permission, [])拿到安全组规则后我的巡检逻辑是逐条检查如果来源是0.0.0.0/0且端口是22、3389、3306、6379、27017这类高危端口直接判fail。如果来源是0.0.0.0/0端口是80或443判warn提示称公网全放通Web端口如果业务不需要请收敛。如果来源是特定IP段且端口不在高危范围内判pass。磁盘使用率这一步不需要直接调API因为磁盘使用率是一个动态指标阿里云监控API提供了查询接口。我用DescribeDiskMonitorData或者直接用云监控的查询接口把磁盘使用率拉下来和阈值做比较。更简单的方式是直接读取Agent上报的数据但这个依赖是否安装了云监控插件所以我倾向于用监控API。3.3 SLB巡检与证书到期提醒SLB巡检有几个重点场景我逐个说。后端服务器健康状态from aliyunsdkslb.request.v20140515.DescribeHealthStatusRequest import DescribeHealthStatusRequest def check_slb_backend_health(client, load_balancer_id): request DescribeHealthStatusRequest.DescribeHealthStatusRequest() request.set_LoadBalancerId(load_balancer_id) response client.do_action_with_exception(request) data json.loads(response) backends data.get(BackendServers, {}).get(BackendServer, []) down_backends [backend for backend in backends if backend.get(ServerHealthStatus) ! normal] return down_backends如果down列表不为空说明有后端ECS健康检查失败需要立即告警。这个巡检项我设置的频率是每30分钟跑一次因为它直接影响线上可用性不能等到第二天早上才发现。证书到期时间这套工具里最有价值的一个巡检项我认为是证书到期提醒。SSL证书过期是一件很麻烦的事浏览器直接拦截用户访问直接报您的连接不是私密连接但证书的到期时间往往被大家忽略。SLB支持挂载证书如果证书是在SLB控制台上传的可以通过DescribeServerCertificates拿到证书列表和到期时间from aliyunsdkslb.request.v20140515.DescribeServerCertificatesRequest import DescribeServerCertificatesRequest def check_cert_expiry(client): request DescribeServerCertificatesRequest.DescribeServerCertificatesRequest() response client.do_action_with_exception(request) data json.loads(response) certs data.get(ServerCertificates, {}).get(ServerCertificate, []) for cert in certs: expire_time datetime.strptime(cert[ExpireTime], %Y-%m-%dT%H:%M:%SZ) days_left (expire_time - datetime.utcnow()).days if days_left 30: # 告警逻辑实践中我发现证书还有一个藏得更深的坑证书绑定的域名和实际业务域名不一致导致某个子域名访问时证书校验失败。这个单纯比对到期时间检查不出来后来我是在巡检项里加了一个检查证书的SANSubject Alternative Name是否包含业务域名的逻辑才把这类问题兜住。3.4 巡检引擎执行流程有了上面的基础能力巡检引擎的执行流程可以梳理成一个完整闭环从InspectTask表取出所有启用的任务。根据任务类型和资源类型从CheckerRegistry取到所有巡检项。从资产表拉取对应的ECS或SLB实例列表。对每个实例逐个执行巡检项。每次执行结果无论pass/fail/warn都写入InspectResult表。如果结果是fail或者warn调用告警模块发通知。4. 巡检项怎么设计才不流于形式4.1 从生产事故中提炼巡检项巡检项的设计决定了这套工具的上限。如果只检查CPU、内存这些表面指标那和云监控没有区别价值不大。我系统梳理了团队在过去一年遇到的所有线上问题把它们分成四类每一类对应一组巡检项第一类容量风险类。磁盘满了是触发频率最高的问题而且它是个渐进过程。今天80%一周之后可能就95%了。这类巡检项的作用是提前预警给你留出处理时间。除了磁盘使用率我还会检查内存使用率、数据盘增长趋势。数据盘增长趋势这个巡检项我专门写了一个逻辑取最近7天的数据按天看增长率如果增长率超过某个阈值比如每天增长2%即使当前绝对数值不高也要告警——因为按这个趋势一个月后就会有问题。第二类配置合规类。安全组规则是否对公网全放通、是否允许密码登录、AccessKey是否长时间未轮换、闲置资源超过30天未启动的ECS是否还在产生费用。这类巡检项的检查频率可以低一些每天一次就够。第三类依赖健康类。SLB后端是否有故障节点、证书是否快到期、域名解析DNS是否指向了正确的服务器、RDS白名单是否包含了不在白名单内的IP。这类巡检项的重要性在于它们的故障症状往往不是直接暴露的是慢慢体现的。比如SLB后端挂了流量会转移到其他后端直到那台也扛不住才会整体雪崩。如果能提前发现后端异常就能及时隔离。第四类资产配置漂移类。所谓配置漂移就是某台ECS的配置被改动了但改的人没走变更流程没人知道。这类巡检项检查的是实际配置是否和期望配置一致。比如我们在Admin里记录了每台ECS的预期用途生产/测试/预发布如果某台测试环境的机器的安全组突然开放了生产环境的端口规则就要告警提醒。配置漂移是巡检工具最值钱的能力之一因为它解决的问题是没人知道你改了什么。4.2 巡检项配置化的思路每个巡检项不能是硬编码的否则换一个团队、换一个场景就得改代码。我在系统里做了一个配置表叫InspectRuleclass InspectRule(models.Model): name models.CharField(max_length128, verbose_name规则名称) resource_type models.CharField(max_length16, verbose_name资源类型) check_item models.CharField(max_length64, verbose_name巡检项代码) threshold models.JSONField(defaultdict, verbose_name阈值配置) enabled models.BooleanField(defaultTrue, verbose_name是否启用) notify_channels models.JSONField(defaultlist, verbose_name通知渠道) class Meta: db_table inspect_rulethreshold字段是一个JSON字典每个巡检项可以定义自己的阈值参数。比如磁盘巡检项的threshold是{ disk_warn_percent: 80, disk_fail_percent: 90, data_disk_growth_percent_per_day: 2 }SLB健康巡检项的threshold是{ max_down_backends: 0 }证书到期巡检项的threshold是{ expire_days_warn: 30, expire_days_fail: 7 }在Admin后台可以直接改阈值不用碰代码。这个设计让我不用每次因为磁盘告警阈值想从80%调到85%就去发布一次代码。注意巡检项的启用/停用也要配置化。有时候某个巡检项写得不完善会产生大量误报如果只能通过改代码去停用会很被动。做成配置化之后先在后台停了等代码修复了再开启避免告警风暴。4.3 巡检结果的评级与聚合巡检结果不应该是简单的pass/fail两态。我给每个巡检结果设置了三个级别pass正常无需处理。warn有隐患但不影响当前业务。比如证书还有20天到期、磁盘使用率到80%、安全组放通了Web端口。这种问题应该进入待办清单但不必半夜打扰人。fail已经产生实际影响或者即将产生不可逆影响。比如SLB健康检查失败、证书7天内到期、磁盘使用率超过90%。在展示层我用了一个实例健康分的概念每个实例的分数是100减去该实例所有巡检项里最低级别对应的扣分。扣分规则是warn扣10分fail扣30分。这样在Admin列表页可以按健康分排序一眼看出哪些机器最需要关注。告警的触发规则也跟级别相关。warn级别只在工作时间9:00-21:00推送fail级别全天推送。这样设计是为了避免狼来了效应——如果半夜三点收到一条warn级别的告警爬起来看一眼发现是安全组放通了443端口那下次收到告警就不会重视了。务必把告警的区分度做出来。5. 定时调度、告警触达与结果可视化5.1 定时调度方案从Celery到APScheduler调度这块我前后用了两套方案走了点弯路分享给你。第一版用的是Celery Beat Redis。Celery很强大支持分布式任务、优先级、路由社区生态也丰富。但它的代价是引入了一整套消息队列体系需要额外维护Redis任务状态在Celery的flower页面里查看数据和Django的ORM是割裂的。对一套企业内部巡检工具来说这些复杂度是多余的。后来我把Celery换成了APScheduler用django-apscheduler插进Django里。APScheduler是一个轻量级的Python定时任务库支持Cron表达式、日期触发、固定间隔触发而且进程内运行不需要额外组件。配置方式非常简单# settings.py INSTALLED_APPS [ ... django_apscheduler, ] # 在App启动时初始化调度器 from apscheduler.schedulers.background import BackgroundScheduler from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job scheduler BackgroundScheduler() scheduler.add_jobstore(DjangoJobStore(), default) def start_scheduler(): scheduler.start()然后每个巡检任务就是一个函数用register_job加上cron触发register_job(scheduler, cron, hour8, minute0, iddaily_ecs_inspect, replace_existingTrue) def daily_ecs_inspect(): # 执行ECS巡检运行方式也很简单Django的command加上一个长驻进程的命令python manage.py runapscheduler用systemd守护这个进程开机自启挂了自动拉起。不需要像Celery那样同时跑worker和beat两个进程运维负担小很多。5.2 告警触达钉钉、飞书、邮件多渠道告警触达是整个系统里最后一公里做得不好前面全白做。我这边按通知渠道分了三个层次第一层次钉钉群机器人。中小企业用钉钉还是飞书的多我这边是钉钉。钉钉群机器人Webhook配置很成熟用Python requests发一个POST请求就行import requests import json def send_dingtalk_alert(webhook_url, title, content): payload { msgtype: markdown, markdown: { title: title, text: content } } response requests.post(webhook_url, jsonpayload) return response.status_code 200Webhook地址保存在系统配置表里支持配置多个群比如运维告警群和开发值班群各一个。第二层次邮件。邮件适合发巡检日报、周报这类汇总信息。用Django自带的send_mail就能实现但要注意配置好Django的SMTP设置EMAIL_BACKEND django.core.mail.backends.smtp.EmailBackend EMAIL_HOST smtp.exmail.qq.com EMAIL_PORT 465 EMAIL_USE_SSL True EMAIL_HOST_USER inspectorexample.com EMAIL_HOST_PASSWORD os.environ.get(MAIL_PASSWORD)第三层次电话/短信。这个我没有直接对接阿里云短信服务而是借道钉钉和飞书的电话通知能力。如果某个巡检项是critical级别比如备用机都挂了钉钉机器人可以配置指定人并触发电话提醒。这样既不增加额外费用也保证了高优先级告警能人工介入。告警降噪是个容易被忽视的点。刚开始跑的时候一个磁盘使用率80%的实例每天的巡检都会发一条warn告警连续发了一周群里全是同一条消息。后来我加了一个连续N次才告警的机制。每个巡检结果落库后会先去查这个实例巡检项最近N次的记录如果最近3次都是warn才真正发告警否则只在界面里展示。等到实际处理完并标记了已恢复之后再发一条恢复通知。这样群里每天看到的都是新问题而不是老问题刷屏。5.3 巡检报表与Django Admin的配合Django Admin是这套工具里被低估的英雄。很多人觉得Admin只是简单的后台管理界面但实际上它自带的列表页、筛选器、搜索、分页、权限控制拿来展示巡检结果几乎不用额外开发。我给InspectResult注册了ModelAdmin之后就获得了一个功能完整的巡检结果查询后台按巡检结果pass/warn/fail筛选按巡检项筛选按实例ID搜索按时间范围筛选排序查看最新结果我还覆写了list_display和list_filteradmin.register(InspectResult) class InspectResultAdmin(admin.ModelAdmin): list_display (instance_id, check_item, result, checked_at) list_filter (result, check_item, checked_at) search_fields (instance_id,) date_hierarchy checked_atdate_hierarchy是Django Admin一个很实用的参数它会在列表页上方生成一个按日期钻取的导航条年-月-日逐级下钻。对于巡检结果这种时间跨度大的数据这个导航条非常实用想查某一天的结果直接点进去就行。报表展示用的是ECharts通过Django的JsonResponse把统计数据输出成JSON然后在前端页面用ECharts画折线图、柱状图。我做了三类报表趋势报表每个巡检项在过去30天的pass率曲线用于观察系统整体健康趋势。实例排行健康分最差的Top 20实例按分数降序排列。告警统计本周共产生多少条warn/fail告警各类型占比。有了报表之后巡检工具就从被动发现问题升级为主动暴露趋势了。6. 在阿里云上的部署实践与运行保障6.1 部署架构一台低配ECS就够很多人以为要部署一套Django数据库调度器得上很多台机器实际上这个工具本身资源消耗非常小我一个2核4G的ECS就够了跑一套Django服务加MySQL绰绰有余。既然是部署在阿里云上直接做了最省事的方案一台2C4G的ECSCentOS 7/Ubuntu 22.04 |- Django APScheduler 主进程 |- MySQL 8.0或者用阿里云RDS看预算 |- Nginx 反向代理 |- Redis没用APScheduler不需要用systemd管理Django进程配了gunicorn[Unit] DescriptionInspector Django Service Afternetwork.target [Service] Typenotify Userwww-data Groupwww-data WorkingDirectory/opt/inspector EnvironmentFile/opt/inspector/.env ExecStart/opt/inspector/venv/bin/gunicorn inspector.wsgi:application --bind 127.0.0.1:8000 --workers 3 --timeout 60 Restartalways RestartSec5 [Install] WantedBymulti-user.target数据库如果用本地MySQL记得定期备份。我给这个库配了一个每天凌晨3点的mysqldump全量备份保留最近7天的备份文件同时备份文件同步到OSS的私有存储桶里。这样即使机器整体挂了也能在半小时内恢复。6.2 数据库选型本地MySQL还是RDS刚开始图省钱用了本地MySQL。跑了三个月一次磁盘扩容操作导致mysqld进程起不来巡检任务全部中断直到第二天才被发现。后来一狠心换成了阿里云RDS MySQL基础版单节点一个月也就一百多块钱换来的是自动备份、自动故障切换、监控告警这些。对巡检工具这种数据量不大但要求稳定的场景RDS的基础版完全够用。表空间预规划方面需要留意。巡检结果表是增长最快的表每天全量跑一遍几十台机器每台十几个巡检项一天就是几千行数据。看似不多但一年下来就是百万行级别。我建议巡检结果表按时间做分区按月建分区表。定期清理超过90天的巡检结果数据或者归档到OSS的CSV文件。分区方案在MySQL 8.0里直接支持ALTER TABLE inspect_result PARTITION BY RANGE (TO_DAYS(checked_at)) ( PARTITION p202501 VALUES LESS THAN (TO_DAYS(2025-02-01)), PARTITION p202502 VALUES LESS THAN (TO_DAYS(2025-03-01)), ... );有了分区表之后删除旧数据只需要DROP PARTITION秒级完成不用DELETE逐条删。6.3 运行时权限与密钥管理整个系统的安全边界也要设计好。第一层Django的SECRET_KEY、数据库密码、阿里云AccessKey等敏感信息全部放在.env文件里由systemd读取代码仓库里只有.env.example模板填的都是假值。第二层绑定RAM子账号的AccessKey只授权读权限。即使泄露了最大损失是信息泄露不至于被删库。第三层Django的ALLOWED_HOSTS只写内网IP或域名不监听公网端口。Nginx监听公网443通过HTTPS转发到后端的Django服务。在Nginx层加了简单的IP白名单只有公司出口IP能访问Admin后台。第四层Django Admin开启了登录二次验证用django-otp插件配合TOTP比如阿里云App或Google Authenticator。这个小工具虽然只是内部系统但管理的是全部云资源的状态信息安全性不能马虎。7. 上线这一年踩过的坑与调优记录7.1 阿里云OpenAPI翻页截断和限流问题前面提到DescribeInstances的翻页问题这是我踩到的第一个坑。第一次写的时候没有处理翻页直接用默认PageSize拉了第一页结果系统里显示的实例数量只有实际的十分之一。巡检结果里安全组开放高危端口这类问题全都漏掉了。好在巡检结果有raw_data字段事后查了下拉取到的数据才发现实例数量对不上。修复方案是写一个通用的分页拉取函数自动判断是否还有下一页def list_all_instances(client, request_class): page_size 100 page_number 1 all_instances [] while True: request request_class() request.set_PageSize(page_size) request.set_PageNumber(page_number) response client.do_action_with_exception(request) data json.loads(response) instances data.get(Instances, {}).get(Instance, []) all_instances.extend(instances) if len(instances) page_size: break page_number 1 return all_instances第二个坑是OpenAPI的限流。阿里云OpenAPI对单账号有QPS限制ECS的DescribeInstances默认QPS是60具体以官方文档为准。但巡检工具每个小时会跑一次实例列表同步加上每个实例的安全组规则查询并发一高就可能触发限流。解决方式是加了一个简单的API调用限速器import time from threading import Lock class ApiRateLimiter: def __init__(self, max_qps20): self.max_qps max_qps self.interval 1.0 / max_qps self.last_call_time 0 self.lock Lock() def wait_if_needed(self): with self.lock: current_time time.time() wait_time self.interval - (current_time - self.last_call_time) if wait_time 0: time.sleep(wait_time) self.last_call_time time.time()Python的time.sleep在毫秒级精度下够了不会阻塞太久。配合了一点随机抖动把整批巡检的QPS控制在20以内再没触发过限流。7.2 时区问题导致定时任务飘移第二坑是时区。Django默认的TIME_ZONE是UTCAPScheduler读取cron表达式的时候也按UTC来算。结果就是我在配置页面写的每天早上8点巡检实际执行时间是北京时间的下午4点。第一天大家没注意第二天看到巡检结果的时间戳全都差了8小时才意识到是时区隔了。修起来也简单settings.py里设置TIME_ZONE Asia/Shanghai USE_TZ FalseUSE_TZ设成False意味着Django直接把本地时间理解为Asia/Shanghai而不是存UTC时间再在展示时转换。这个设置对内部工具来说更直观开发也省心。如果你有多地域需求再考虑USE_TZTrue但一套纯国内系统没必要给自己找麻烦。调度器那边也要同步设置scheduler BackgroundScheduler(timezoneAsia/Shanghai)7.3 慢查询和巡检结果表膨胀第三个坑出现在上线第三个月。巡检结果表数据越来越多了Django Admin页面的列表越来越慢点开一个实例的历史记录要等好几秒。我查了下慢查询日志发现所有查询都在对checked_at做范围扫描但索引只建在instance_id check_item checked_at上没有单独覆盖checked_at的索引。后来做两处优化第一把索引改成了models.Index(fields[instance_id, check_item, checked_at]), models.Index(fields[result, checked_at]),第二把巡检结果表按月分区的Django模型加了一个清理任务每天凌晨2点删除超过90天的记录def cleanup_old_results(): cutoff timezone.now() - timedelta(days90) InspectResult.objects.filter(checked_at__ltcutoff).delete()注意delete大批量数据在MySQL里是逐行删的很容易锁表。我用的是MySQL直接删分区的方式或者分批delete每次删5000条间隙sleep一下。7.4 误报让人麻木必须做告警降噪这是最重要的一课。工具刚上线那会儿每天群里能刷几十条告警大部分是warn级别的安全组放通了443端口磁盘使用率到80%。开发群的人一开始还认真看一眼后来就麻木了把钉钉群设了免打扰。然后真正出问题的时候告警虽然发了但没人在意直接在免打扰的通知列表里沉底了。我花了两周时间重构告警策略走了两条路一是降低warn级告警的推送频率。warn只在工作日早上9点到晚上9点推送且同一个实例同一个巡检项一天最多推一次。fail级的告警不受这个限制随时推。二是允许在Admin后台把某个巡检项标记为已知问题。比如某台ECS因为业务原因就是要对外开放22端口做远程调试运维确认过不需要处理就可以在后台给这个实例的该巡检项加一个白名单下次巡检直接跳过。白名单里记录了解除的时间和负责人做到可审计。做了这两件事之后告警量降到之前的五分之一告警的有效性大幅提升。7.5 巡检项误伤业务的处理白名单与例外机制巡检工具本质上是用规则去卡所有机器。但线上环境总有特殊的机器它们因为业务原因就是不符合通用规则。比如某台机器需要在安全组上公网放通3306端口给合作伙伴联调数据库本来就是给第三方系统同步用的。这种场景如果巡检工具不区分每跑一次就告警一次最终就会被运维忽略。我的处理方式是给每个巡检结果加了一个白名单表class InspectWhitelist(models.Model): instance_id models.CharField(max_length64, verbose_name实例ID) check_item models.CharField(max_length64, verbose_name巡检项代码) reason models.CharField(max_length256, verbose_name白名单原因) created_by models.CharField(max_length32, verbose_name操作人) created_at models.DateTimeField(auto_now_addTrue) class Meta: db_table inspect_whitelist unique_together (instance_id, check_item)在执行引擎里每跑一个巡检项之前先查一下如果这个实例在这个巡检项上有白名单记录就跳过。白名单的添加必须在Admin后台操作带操作人信息这样每一项破例都是可追溯的。这个机制上线后误报问题才算从根上解决了。工具做出来之后我们日常巡检的负担小了很多。以前是每天早上运维到控制台一台一台点开看现在只需要看一眼钉钉群里有没有fail级别的告警没有就安心有就点开巡检平台定位处理。这段时间持续积累下来我看到最大的收获其实不是省了多少工作量而是团队对线上配置的信心不一样了——以前谁也不确定安全组规则有没有被谁改过证书哪天到期也没人记得现在这些问题有了一个兜底的答案。如果你也管着批云上资产又觉得云监控不够用不妨参照这个思路自己搭一套不用一次做全先挑一个最痛的点开始跑起来之后再慢慢加巡检项。本文还有配套的精品资源点击获取