★ FEATURED ARTICLE 简介这份资源是一套基于Django框架搭建的运维平台源码面向具备Python基础、希望了解Web运维系统开发的IT从业者与学习者。平台采用前后端不分离架构将主机管理、服务监控、日志查看、任务调度、权限控制、资源统计、脚本执行、故障告警、审计追踪与知识库等功能整合于同一项目适合作为小型运维工具原型或Django实战练手项目。压缩包共164个文件约3.63MB以39个py源码文件为核心配合38个js、23个css及11个html构成前端界面另有png、svg、ttf等图片与字体资源以及少量日志、配置与说明文档目录结构完整便于按模块阅读。目前已有2716人学习下载。通过研读模型定义、视图函数与模板文件读者可掌握Django项目组织方式、后台管理界面定制及前后端协作思路并理解Linux运维场景下的功能设计逻辑为后续二次开发或搭建自有运维平台提供可复用的参考。1. 从一堆散落的脚本说起这套运维平台到底要解决什么很多团队早期的运维状态是这样的十几台机器每台上面散落着几十个 shell 脚本部署靠手动 scp重启服务靠 SSH 上去敲命令日志要一台台登录去看。人少的时候还能扛一旦机器数量上到三五十台或者来了新人接手这套东西立刻变成黑匣子——没人说得清哪台机器上跑的是什么版本也没人敢随便动那些脚本。这个项目要解决的就是把这种「脚本游击队」状态收敛成一个统一的 Web 运维平台用 Django 做后端把主机管理、任务下发、部署发布、日志查看这几件事收进一个界面里。它适合两类人一类是正在做 Python 方向毕设或课程设计、需要一套结构完整可交付系统的学生另一类是小团队里想自己搭一套轻量运维工具的开发者不想上重型平台但又受够了手工操作。2. 需求拆解与技术选型为什么是 Django 而不是 Flask2.1 先把功能边界划清楚做这类平台最容易翻车的地方是一开始什么都想要最后什么都没做完。我一般会把需求压到四个核心模块其余全部砍掉或放到二期主机管理维护被管机器的 IP、SSH 端口、登录凭据、分组标签支持增删改查和批量导入。任务执行选一批主机下发一条命令或一段脚本实时或准实时拿到每台机器的执行结果。应用部署把「拉代码 → 装依赖 → 重启服务」这套流程固化成模板一键对指定主机组执行。操作审计谁在什么时间对哪些机器做了什么操作全部落库出问题能回溯。这四个模块覆盖了日常运维 80% 的高频动作剩下的监控告警、CMDB 自动发现这些属于锦上添花毕设或课设阶段不建议铺开。2.2 技术选型Django 的取舍选 Django 而不是 Flask核心理由是这类平台天然需要一套完整的后台能力用户认证、权限控制、ORM、Admin 后台、表单校验Django 全都自带。用 Flask 你得自己拼 SQLAlchemy、Flask-Login、WTForms拼完发现工作量比 Django 还大而且权限模型要自己设计。Django 的 Admin 还能在开发阶段直接当临时管理界面用省掉大量前端联调时间。具体技术栈我一般这样定层次选型说明Web 框架Django 4.x自带 ORM、认证、Admin异步任务Celery Redis命令下发不能阻塞请求远程执行Paramiko纯 Python 实现 SSH无需装客户端前端Bootstrap 少量 jQuery不引入重型前端框架降低门槛数据库MySQL 或 PostgreSQL生产用 PG课设用 MySQL 也行部署Gunicorn Nginx标准组合这里有个关键决策命令执行必须异步化。如果直接在 Django 视图里同步跑 Paramiko一条命令卡 30 秒整个请求就挂 30 秒并发一上来直接崩。所以用 Celery 把执行任务丢到队列里视图只负责创建任务并返回任务 ID前端轮询或走 WebSocket 拿结果。提示课设环境如果不想装 Redis可以用 Django 自带的django.core.mail那种思路把 Celery 换成线程池但生产环境不要这么干线程池扛不住并发也管不了任务状态。2.3 数据模型设计模型设计直接决定后面写起来顺不顺。核心表我一般这样建# hosts/models.py from django.db import models class HostGroup(models.Model): 主机分组用于批量操作 name models.CharField(max_length64, uniqueTrue) description models.TextField(blankTrue) def __str__(self): return self.name class Host(models.Model): 被管主机 hostname models.CharField(max_length128) ip models.GenericIPAddressField() ssh_port models.IntegerField(default22) username models.CharField(max_length64) # 密码和密钥不存明文存加密后的密文 credential models.TextField(help_text加密后的登录凭据) group models.ForeignKey(HostGroup, on_deletemodels.SET_NULL, nullTrue, related_namehosts) is_active models.BooleanField(defaultTrue) created_at models.DateTimeField(auto_now_addTrue) class Meta: unique_together (ip, ssh_port) class TaskRecord(models.Model): 任务执行记录审计用 STATUS_CHOICES [ (pending, 待执行), (running, 执行中), (success, 成功), (failed, 失败), ] operator models.CharField(max_length64) command models.TextField() status models.CharField(max_length16, choicesSTATUS_CHOICES, defaultpending) created_at models.DateTimeField(auto_now_addTrue) finished_at models.DateTimeField(nullTrue, blankTrue)Host表里凭据字段用TextField存密文加密密钥放环境变量不要写死在代码里。TaskRecord是审计的核心每次下发命令都落一条operator字段记录操作人。unique_together防止同一台机器重复录入。3. 核心模块实现从 SSH 连接到任务下发3.1 用 Paramiko 封装一个可靠的执行器Paramiko 用起来简单但坑不少。最典型的是连接超时和命令执行超时不分导致一个卡死的命令把整个 worker 拖住。我一般会封装一个执行器类把连接和执行分开控制# executor/ssh_client.py import paramiko from paramiko import SSHClient, AutoAddPolicy class SSHExecutor: def __init__(self, ip, port, username, passwordNone, keyNone, connect_timeout10, exec_timeout60): self.ip ip self.port port self.username username self.password password self.key key self.connect_timeout connect_timeout self.exec_timeout exec_timeout def run(self, command): 执行单条命令返回 (exit_code, stdout, stderr) client SSHClient() # 生产环境应加载 known_hosts 做校验课设可先用 AutoAddPolicy client.set_missing_host_key_policy(AutoAddPolicy()) try: client.connect( hostnameself.ip, portself.port, usernameself.username, passwordself.password, key_filenameself.key, timeoutself.connect_timeout, banner_timeoutself.connect_timeout, ) stdin, stdout, stderr client.exec_command( command, timeoutself.exec_timeout ) exit_code stdout.channel.recv_exit_status() out stdout.read().decode(utf-8, errorsignore) err stderr.read().decode(utf-8, errorsignore) return exit_code, out, err except Exception as e: return -1, , str(e) finally: client.close()几个参数要重点说connect_timeout控制 TCP 握手和认证阶段exec_timeout控制命令执行阶段两者分开设。recv_exit_status()会阻塞直到命令结束所以exec_timeout必须设否则一个tail -f就能把 worker 挂死。errorsignore是防止某些机器返回非 UTF-8 编码导致解码报错。3.2 Celery 任务把执行结果落库执行器写好了接下来是把它接到 Celery 里。任务函数要做三件事更新任务状态为 running、逐台执行、把结果写回数据库。# executor/tasks.py from celery import shared_task from django.utils import timezone from .models import TaskRecord, Host from .ssh_client import SSHExecutor shared_task(bindTrue, max_retries2) def run_command_task(self, task_id, host_ids, command): task TaskRecord.objects.get(idtask_id) task.status running task.save(update_fields[status]) results [] for host in Host.objects.filter(id__inhost_ids, is_activeTrue): executor SSHExecutor( iphost.ip, porthost.ssh_port, usernamehost.username, passworddecrypt(host.credential), ) code, out, err executor.run(command) results.append({ host: host.hostname, ip: host.ip, exit_code: code, stdout: out[-2000:], # 只留尾部防止结果过大 stderr: err[-2000:], }) task.status success if all(r[exit_code] 0 for r in results) else failed task.finished_at timezone.now() task.save(update_fields[status, finished_at]) return resultsbindTrue让任务能访问self方便做重试。max_retries2是防止网络抖动导致任务直接失败。out[-2000:]截断输出因为有些命令输出几十兆全存进数据库会把表撑爆。decrypt是自定义的解密函数和存凭据时的加密函数对应。3.3 部署模板把流程固化下来部署模块的本质是把一串命令按顺序执行中间任何一步失败就中断。我一般用一个简单的模板表存命令列表执行时按顺序跑# deploy/models.py class DeployTemplate(models.Model): name models.CharField(max_length128) # 每行一条命令按顺序执行 steps models.TextField(help_text每行一条命令) target_group models.ForeignKey(hosts.HostGroup, on_deletemodels.CASCADE) def get_steps(self): return [s.strip() for s in self.steps.splitlines() if s.strip()]执行时遍历get_steps()每条命令调一次SSHExecutor.runexit_code ! 0就 break 并记录失败步骤。这样设计的好处是模板可编辑、可复用新人不用懂脚本也能操作。4. 避坑与常见问题那些年踩过的 SSH 和 Celery 的坑4.1 现象任务一直卡在 running永远不结束原因最常见的是 Paramiko 的exec_timeout没设或者设了但命令本身在等输入比如sudo要密码、read等用户输入。另一种是 Celery worker 数量不够任务排队但没 worker 消费。解决所有exec_command必须带timeout参数命令里避免交互式操作sudo用-S从 stdin 读密码或配免密Celery 启动时用-c指定并发数监控队列长度积压了就加 worker。4.2 现象部分机器连接成功但命令返回空原因目标机器的 shell 环境变量没加载。Paramiko 执行命令时默认不走 login shell/etc/profile里的 PATH 不生效导致java、node这些命令找不到。解决命令前加source /etc/profile 或者用bash -l -c your command走 login shell。我一般直接在执行器里包一层统一处理。4.3 现象Celery 任务重复执行原因acks_late配置和 worker 崩溃的组合。默认 Celery 在任务开始前就 ackworker 崩了任务丢失开了acks_late后任务执行完才 ack但 worker 执行到一半崩了任务会被重新投递导致重复执行。解决任务函数必须幂等。比如部署任务重复执行前先检查当前版本已经是目标版本就跳过。或者用 Redis 锁任务开始时抢锁抢不到直接返回。4.4 现象凭据加密后解密失败原因加密密钥换了或者加密时用的库版本和解密时不一致。Fernet 这类库对密钥格式敏感密钥变了旧数据全解不开。解决密钥放环境变量并做好备份不要硬编码。加密库锁定版本升级前先测试解密旧数据。生产环境建议引入密钥管理服务课设阶段至少把密钥写进.env并加入.gitignore。4.5 现象批量执行时部分机器超时拖垮整体原因串行执行一台机器卡 60 秒十台就是 10 分钟。用户等不及刷新页面以为系统挂了。解决用 Celery 的group或chord做并行执行每台机器一个子任务最后汇总结果。并行度受 worker 数量限制但比串行快得多。前端用轮询任务状态不要同步等待。5. 验收与答辩怎么证明这套平台真的能用验收这件事很多人栽在「功能演示很流畅一问细节就露馅」。我一般会准备三层验证从功能到边界到异常逐层过。第一层功能闭环验证。准备三台测试机虚拟机就行录入平台建一个分组。然后走一遍完整流程下发uptime命令确认三台都返回结果建一个部署模板内容是echo deploy step 1和echo deploy step 2执行后确认两步都成功最后查审计记录确认操作人、时间、命令都对得上。这一层证明系统能跑通。第二层边界与异常验证。这是答辩时拉开差距的地方。故意关掉一台测试机的 SSH 服务再下发命令看平台是否优雅地返回「连接失败」而不是抛 500故意下发一条sleep 120的命令看是否在exec_timeout后超时返回故意下发一条不存在的命令看exit_code是否非零且 stderr 有内容。把这些异常场景的结果截图放进答辩 PPT比只演示成功路径有说服力得多。第三层性能与并发验证。如果时间允许用脚本模拟 20 台机器同时下发命令观察 Celery 队列和任务完成时间。不需要做到生产级压测但至少证明「批量操作不会把系统卡死」。可以用locust或简单的 Python 脚本发请求。答辩时还有一个高频问题「你的平台和 Ansible 有什么区别」这个问题不要回避直接说定位不同——Ansible 是命令行工具面向熟练运维这套平台是 Web 界面面向团队协作和审计把操作门槛降下来同时留下操作记录。两者不冲突平台底层甚至可以调 Ansible 做执行引擎只是这个项目为了教学完整性自己实现了 SSH 层。注意答辩演示前一定要把测试环境跑一遍尤其是 Celery 和 Redis 是否启动、测试机是否可达。我见过太多现场演示时 Redis 没起来、任务一直 pending 的翻车现场。从那以后我每次做这类平台都会在验收前强制走一遍「关服务、断网络、发错命令」这三步异常测试确认系统不会崩、不会卡、不会丢数据。这套习惯帮我省掉了无数次答辩现场的尴尬。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
