简介：这份资源是一套基于Django框架开发的运维平台源码，面向具备Python基础、希望了解Web运维系统实现思路的开发者与运维人员，可用于学习主机管理、服务监控、日志查看、任务调度、权限控制、资源统计、脚本执行与故障告警等常见运维模块的落地方式。压缩包共164个文件，以39个py源码文件为核心，配合38个js、23个css、11个html及字体图标等前端资源，另有少量日志、配置与说明文件，整体约3.63MB，采用前后端不分离架构，目录结构完整，便于按模块阅读与二次开发。目前已有2718人学习下载。读者可通过运行项目、梳理模型与视图逻辑，理解Django在运维场景中的后台管理与数据建模用法，并借鉴其功能划分与页面组织方式，作为课程设计或小型运维工具的参考原型。
1. 运维平台不是监控大屏：从一堆散落脚本到可交付的 Django 运维系统
我见过太多团队把「运维平台」做成了首页挂几个折线图、点进去全是静态截图的面子工程。真正卡住日常工作的，从来不是缺一块大屏，而是发布要手动登机器、巡检靠人肉敲命令、故障排查时日志散在七八个地方。这套基于 Django 的运维平台，目标很明确：把资产、发布、巡检、告警这四件高频事收进一个 Web 入口，让值班的人不用记一堆脚本路径。它适合两类人——要交毕设或课程设计、需要一套结构完整可讲清的系统；以及小团队里想把零散 shell 脚本沉淀成工具链的开发者。Python 开发在 DevOps 工具这个方向上的优势，恰恰是胶水能力强、上手快，Django 自带 ORM、Admin 和权限体系，能省掉大量脚手架工作。
2. 需求与选型：四类角色、五个核心模块的边界划分
动手写代码之前，先把「谁用、用来干什么」定死，否则后面每加一个功能都要返工。运维平台的角色通常分四类：管理员负责资产录入和权限分配，运维工程师执行发布和巡检，开发人员只读查看自己服务的状态，值班人员处理告警。这四类人对界面的诉求完全不同，权限模型必须从第一天就设计进去，而不是最后补一个is_superuser判断。
2.1 功能边界：先砍掉「什么都想要」
一个能交付的运维平台，核心模块控制在五个：资产管理（主机、分组、标签）、任务执行（批量命令、脚本下发）、发布管理（应用、版本、发布单）、巡检与告警（定时任务、阈值规则、通知）、审计日志（谁在什么时候对哪台机器做了什么）。这五块覆盖了日常 80% 的操作。像 CMDB 自动发现、拓扑图、容量预测这类，属于加分项，毕设里可以作为扩展章节提一句，但不要放进主线，否则工期一定崩。
我一般会画一张模块职责表，把每个模块的输入输出写清楚，这张表后面直接变成数据库设计和接口设计：
| 模块 | 核心输入 | 核心输出 | 依赖 |
|---|---|---|---|
| 资产管理 | 主机 IP、分组、标签、凭据引用 | 资产列表、分组树 | 无 |
| 任务执行 | 目标主机列表、命令/脚本、超时 | 执行结果、退出码、耗时 | 资产管理 |
| 发布管理 | 应用、版本包、目标环境 | 发布单状态、回滚记录 | 资产管理、任务执行 |
| 巡检告警 | 巡检项、阈值、周期 | 巡检报告、告警事件 | 任务执行 |
| 审计日志 | 所有写操作 | 操作流水 | 全模块 |
这张表的价值在于：它逼你想清楚「任务执行」是发布和巡检的公共底座，所以它必须先做、做稳，否则上面两个模块都是空中楼阁。
2.2 技术选型：为什么是 Django 而不是 Flask 或 FastAPI
选 Django 的核心理由是「交付速度」。这个平台本质是一个带权限、带后台、带表单的 CRUD 密集型系统，Django 的 Admin 能让你在半天内把资产录入界面跑起来，ORM 的迁移机制让表结构演进有据可查，内置的认证和权限组直接对应四类角色。Flask 更轻但什么都要自己拼，FastAPI 异步性能好但这个场景并发并不高、反而要自己搭后台和权限。
技术栈我通常这样定：Django 4.x + Django REST Framework 做前后端分离的接口层，前端用简单的模板渲染或轻量框架都行，数据库 MySQL 或 PostgreSQL，异步任务用 Celery + Redis 处理批量执行和定时巡检，远程执行走 SSH 协议（paramiko 或 ansible 的 Python API）。这里有个血泪经验：批量执行千万别放在 Web 请求的同步流程里，几十台机器串行 SSH 能把请求拖到超时，必须丢给 Celery 异步跑，前端轮询任务状态。
# settings.py 关键配置片段 INSTALLED_APPS = [ "django.contrib.admin", "django.contrib.auth", "rest_framework", "apps.assets", # 资产管理 "apps.executor", # 任务执行 "apps.release", # 发布管理 "apps.inspect", # 巡检告警 "apps.audit", # 审计日志 ] # Celery 配置，任务结果写回 Redis CELERY_BROKER_URL = "redis://127.0.0.1:6379/0" CELERY_RESULT_BACKEND = "redis://127.0.0.1:6379/1" CELERY_TASK_TRACK_STARTED = True # 记录任务开始时间，前端可展示进度 CELERY_TASK_TIME_LIMIT = 600 # 单任务硬超时，防止卡死 worker这段配置里CELERY_TASK_TIME_LIMIT是很多人会漏的，批量任务一旦某台机器 SSH 卡住，没有硬超时会把 worker 占满，后面所有任务排队，表现就是「平台突然不动了」。TRACK_STARTED打开后前端能区分「排队中」和「执行中」，体验差别很大。
2.3 权限模型：用 Django 权限组映射四类角色
不要自己造权限表，Django 的Group+Permission够用。做法是建四个组，把模型级权限挂上去，视图里用@permission_required或 DRF 的permission_classes控制。开发人员只给view权限，运维给add/change，管理员给全部。这样审计日志里记录的用户操作天然带角色上下文，答辩时讲权限设计也有据可依。
3. 总体架构与数据模型：从请求入口到执行落地的链路
架构讲不清，答辩第一个问题就会卡住。这套平台的请求链路是：浏览器 → Django 视图/DRF 接口 → 业务逻辑层 → Celery 任务队列 → 执行器（SSH/Ansible）→ 目标主机，结果回写数据库，前端轮询或 WebSocket 拿状态。分层的关键是「视图只做参数校验和权限判断，业务逻辑放 service 层，执行细节放 executor」，这样同一套执行逻辑能被发布和巡检复用。
3.1 数据模型：五张核心表撑起整个平台
资产、任务、发布单、巡检项、审计日志这五张表是骨架。下面给出资产和任务两张关键表的定义，其余按同样思路扩展：
# apps/assets/models.py from django.db import models class HostGroup(models.Model): name = models.CharField(max_length=64, unique=True) parent = models.ForeignKey("self", null=True, blank=True, on_delete=models.CASCADE, related_name="children") class Host(models.Model): ip = models.GenericIPAddressField(unique=True) hostname = models.CharField(max_length=128) group = models.ForeignKey(HostGroup, on_delete=models.PROTECT) tags = models.JSONField(default=list) # 如 ["prod", "web"] credential_ref = models.CharField(max_length=128) # 凭据在密钥库中的引用，不存明文 is_active = models.BooleanField(default=True) created_at = models.DateTimeField(auto_now_add=True)credential_ref存的是引用而不是密码明文，这是安全底线，答辩时被问到「密码怎么存的」能直接答上来。group用PROTECT而不是CASCADE，防止误删分组把主机一起带走，这个细节体现的是对生产数据的敬畏。
# apps/executor/models.py class ExecTask(models.Model): STATUS = [("pending", "排队"), ("running", "执行中"), ("success", "成功"), ("failed", "失败")] name = models.CharField(max_length=128) command = models.TextField() targets = models.ManyToManyField("assets.Host") status = models.CharField(max_length=16, choices=STATUS, default="pending") created_by = models.ForeignKey("auth.User", on_delete=models.PROTECT) created_at = models.DateTimeField(auto_now_add=True) class ExecResult(models.Model): task = models.ForeignKey(ExecTask, on_delete=models.CASCADE, related_name="results") host = models.ForeignKey("assets.Host", on_delete=models.CASCADE) exit_code = models.IntegerField(null=True) stdout = models.TextField(blank=True) stderr = models.TextField(blank=True) duration_ms = models.IntegerField(default=0)任务和结果拆两张表，是因为一次任务对多台主机，结果是一对多关系。duration_ms单独存，后面做「哪台机器执行慢」的分析直接聚合这个字段，不用去解析日志。
3.2 执行器封装：一个函数吃下所有远程执行
执行器是整个平台的心脏，封装得好，发布和巡检都省事。核心是一个「给定主机和命令，返回退出码和输出」的函数，内部处理连接超时、命令超时、编码问题：
# apps/executor/runner.py import paramiko from django.conf import settings def run_on_host(host, command, timeout=30): """在单台主机执行命令，返回 (exit_code, stdout, stderr)""" client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect( hostname=host.ip, username=settings.SSH_USER, key_filename=settings.SSH_KEY_PATH, # 统一用密钥，不用密码 timeout=10, banner_timeout=10, ) stdin, stdout, stderr = client.exec_command(command, timeout=timeout) exit_code = stdout.channel.recv_exit_status() return (exit_code, stdout.read().decode("utf-8", errors="replace"), stderr.read().decode("utf-8", errors="replace")) except Exception as exc: return (-1, "", f"connection error: {exc}") finally: client.close()三个参数值得说：timeout=10是连接超时，机器失联时快速失败而不是干等；exec_command的timeout是命令执行超时，防止tail -f这类命令把连接挂死；errors="replace"处理非 UTF-8 输出，否则中文日志会直接抛异常。返回-1表示连接层失败，和命令本身返回非零要区分开，前端展示时一个是「连不上」一个是「命令报错」，排查方向完全不同。
3.3 异步任务编排：Celery 任务怎么写才不翻车
批量执行任务用 Celery 的group并发下发，每台主机一个子任务，主任务汇总结果：
# apps/executor/tasks.py from celery import shared_task, group @shared_task def exec_on_single_host(host_id, command, result_id): from apps.assets.models import Host from apps.executor.models import ExecResult from apps.executor.runner import run_on_host host = Host.objects.get(id=host_id) code, out, err = run_on_host(host, command) ExecResult.objects.filter(id=result_id).update( exit_code=code, stdout=out, stderr=err) @shared_task def dispatch_task(task_id): from apps.executor.models import ExecTask, ExecResult task = ExecTask.objects.get(id=task_id) task.status = "running" task.save(update_fields=["status"]) jobs = [] for host in task.targets.all(): result = ExecResult.objects.create(task=task, host=host) jobs.append(exec_on_single_host.s(host.id, task.command, result.id)) group(jobs).apply_async()这里有个容易踩的坑：ExecResult必须先创建再传 id 给子任务，因为子任务在独立进程里跑，不能直接传 ORM 对象。另外dispatch_task里不要select_related后把对象序列化传参，Celery 序列化 ORM 对象会出问题，传 id 最稳。
4. 核心模块实现：发布、巡检、告警三条主线的落地细节
模块实现阶段最容易失控的是「发布管理」，因为它牵扯版本包、目标环境、回滚，逻辑分支多。巡检和告警相对线性，但定时任务的可靠性是重点。这一章按三条主线拆开讲，每条都给到能直接改的代码骨架。
4.1 发布管理：发布单状态机与回滚设计
发布单本质是一个状态机：待审批 → 发布中 → 成功/失败 → 已回滚。状态流转必须用数据库事务包住，防止并发操作把状态改乱。核心逻辑是「先备份当前版本，再拉新包，再重启服务，失败自动回滚」：
# apps/release/services.py from django.db import transaction from apps.executor.runner import run_on_host RELEASE_STEPS = [ "mkdir -p /opt/app/releases/{version}", "cp -r /opt/app/current/* /opt/app/backup/", # 备份当前版本 "tar -xzf /tmp/{package} -C /opt/app/releases/{version}", "ln -sfn /opt/app/releases/{version} /opt/app/current", "systemctl restart myapp", ] def deploy(release, hosts): with transaction.atomic(): release.status = "deploying" release.save(update_fields=["status"]) for host in hosts: for step in RELEASE_STEPS: cmd = step.format(version=release.version, package=release.package) code, out, err = run_on_host(host, cmd) if code != 0: rollback(host) # 任一步失败立即回滚该主机 release.status = "failed" release.save(update_fields=["status"]) return False release.status = "success" release.save(update_fields=["status"]) return True def rollback(host): run_on_host(host, "ln -sfn /opt/app/backup /opt/app/current && systemctl restart myapp")ln -sfn做软链切换是发布的核心技巧，切换是原子的，比直接覆盖文件安全得多。回滚就是把软链指回备份目录再重启，秒级完成。注意transaction.atomic()只包状态更新，不包远程执行——远程操作不可回滚，放进事务里没有意义还会长时间锁行。
4.2 巡检模块：巡检项配置化，别把检查逻辑写死
巡检最忌讳把检查项写死在代码里，加一个检查就要改代码发版。正确做法是巡检项存数据库，用「命令 + 判定规则」描述：
# apps/inspect/models.py class InspectItem(models.Model): name = models.CharField(max_length=64) # 如「磁盘使用率」 command = models.TextField() # 采集命令 rule_type = models.CharField(max_length=16) # gt / lt / contains threshold = models.CharField(max_length=64) # 阈值 enabled = models.BooleanField(default=True)判定逻辑用一个分发函数处理，gt表示采集值大于阈值即异常：
def judge(rule_type, threshold, output): try: value = float(output.strip()) except ValueError: return rule_type == "contains" and threshold in output if rule_type == "gt": return value > float(threshold) if rule_type == "lt": return value < float(threshold) return False这样运维人员自己在后台加一条「内存使用率 > 90」的巡检项，不用找开发。巡检结果按批次存报告，前端按主机维度展示红黄绿，值班的人一眼能看出哪台机器有问题。
4.3 告警通知：去重、抑制和升级三件事
告警模块的难点不在发通知，而在「别把人吵死」。三个机制必须有：去重（同一主机同一巡检项在冷却期内只发一次）、抑制（主机整体失联时，抑制它上面所有单项告警）、升级（P1 告警 15 分钟未处理自动升级通知上级）。去重用一个 Redis key 带过期时间实现最简单：
import redis from django.conf import settings r = redis.Redis.from_url(settings.CELERY_BROKER_URL) def should_notify(host_id, item_id, cooldown=1800): key = f"alert:{host_id}:{item_id}" # set nx ex 原子操作：不存在才写入并设过期，返回 True 表示首次告警 return bool(r.set(key, "1", nx=True, ex=cooldown))nx=True保证原子性，多进程并发也不会重复发。冷却期 1800 秒按业务调，核心服务可以短一些，边缘服务长一些。通知渠道先接一个（邮件或企业微信机器人），别一上来接五个，维护成本吃不消。
5. 避坑与排查：上线后最容易翻车的五个点
这一章是我踩过的坑攒出来的，每一条都按「现象 → 原因 → 解决」写，照着排查能省不少时间。
现象一：批量执行任务一直「排队中」，平台像卡死。原因通常是 Celery worker 没起、或者 worker 被前一个卡住的任务占满。先看celery -A proj status有没有活跃 worker，再看CELERY_TASK_TIME_LIMIT是否生效。解决是给每个远程执行加硬超时，worker 启动参数加--max-tasks-per-child=50定期回收，防止内存泄漏累积。
现象二：SSH 连接偶发失败，重试就好。原因是目标机MaxStartups限制或网络抖动，批量并发太高时尤其明显。解决是给 Celery 任务加autoretry_for=(Exception,)和retry_backoff=True，同时控制并发数（worker 的--concurrency别超过目标机承受能力），别一次对 200 台机器同时开 SSH。
现象三：发布后服务起不来，但发布单显示成功。原因是systemctl restart返回 0 不代表服务真的健康，进程可能起来又挂了。解决是在发布步骤最后加一个健康检查，比如curl -f http://127.0.0.1:8000/health连续成功三次才算发布成功，否则触发回滚。这个检查是发布流程的「后悔药」，一定要有。
现象四：审计日志查不到某次操作。原因是写操作走了 DRF 的perform_create但没统一埋点，漏了某个接口。解决是用 Django 的signals（post_save/post_delete）统一记录，或者在 DRF 的permission层加中间件统一写日志，别指望每个视图手动写。
现象五：巡检报告数据对不上，同一台机器两次结果差异大。原因是巡检命令本身有缓存或采样窗口，比如free和df在不同时刻取值不同。解决是巡检项里明确采集命令的采样方式（如df -P固定格式），并在报告里记录采集时间戳，对比时按时间对齐，别拿两次不同时刻的数据硬比。
6. 验收与答辩：用一条发布链路把整套系统讲透
验收阶段最忌讳从头到尾念功能列表。评委或面试官真正想确认的是「这套系统能不能跑通一条完整链路」。我的建议是准备一条演示主线：从资产录入一台主机开始，创建一个发布单，触发发布，中途故意让某一步失败，展示自动回滚，最后在审计日志里看到完整操作记录。这一条链路串起了资产、执行、发布、审计四个模块，比逐个模块点一遍有说服力得多。
验收标准可以量化成一张表，答辩时直接展示：
| 验收项 | 合格标准 | 验证方式 |
|---|---|---|
| 资产录入 | 支持分组、标签、批量导入 | 导入 10 台主机，分组正确 |
| 批量执行 | 50 台以内 2 分钟内完成 | 计时执行uptime |
| 发布回滚 | 失败后 30 秒内恢复上一版本 | 故意改错包路径触发 |
| 巡检告警 | 阈值触发后 1 分钟内通知 | 手动造高负载 |
| 审计完整 | 所有写操作可追溯 | 抽查 5 次操作 |
答辩时被问到「你这个和现成的开源工具比有什么价值」，别硬吹功能多，就讲「这套是按我们自己的发布流程定制的，状态机和回滚策略贴合实际，而且代码量可控、能自己维护」。诚实讲边界，比夸大更加分。
还有一个技巧：提前把演示环境的数据准备好，但留一台「故意有问题」的主机，现场演示告警和排查。评委看到系统能发现问题、能定位，比看到一切正常更信服。我一般会在演示前把整条链路自己走三遍，每遍都掐表，把耗时和可能的卡点记下来——从那以后我每次交方案前都强制走一遍「失败路径」，因为成功路径谁都会演示，失败路径才见真功夫。希望这套结构能帮到你，把方案稳稳落地。
本文还有配套的精品资源，点击获取
