目录
需求
功能介绍
页面效果
代码编写
docker部署
需求
线上主机一百台左右,经常会在某个服务器上放置一些自动化脚本,并配置计划任务,时间长可能忘记计划任务所在服务器,所以开发一个用于收集展示crontab任务的服务
语言框架
python3.9 / Django
模块:Django~=4.2.3、paramiko~=3.2.0
版本别差太多就行
功能介绍
具有收集功能,并将收集到的信息写入到文件中,用于存储
打开主页会获取文件中的任务信息,并渲染到页面
首次打开主页并不会获取到任务信息,需要点击‘重新获取’按钮进行任务获取和存储
当服务器增加或减少任务,可点击‘重新获取’来更新最新的任务
页面效果
代码编写
编译器 pycharm 创建django服务
服务目录结构:
crontab/urls.py
from django.contrib import admin
from django.urls import path
from host_cron import views as crontab
urlpatterns = [
path('admin/', admin.site.urls),
path('crontab_get/', crontab.crontab_get, name='crontab_get'),
path('', crontab.crontab_select)
]
crontab/config.py
该文件用于保存主机列表、登陆用户名、登陆密钥、存储任务信息的文件路径
因这些配置可能会根据不同环境变化,所以该文件在用docker启动时采用外挂的方式
def Host_Add():
'''
用于crontab views中 crontab_get调用
:return: 主机ip列表
'''
hostname_list = ['172.100.0.2',
'172.100.0.3']
return hostname_list
def Host_User():
'''
用于crontab views中 crontab_get调用
:return: 主机登陆用户名
'''
user = 'root'
return user
def Host_Key_Path():
'''
用于crontab views中 crontab_get调用
:return: 主机登陆密钥路径
'''
key_path = '/data/id_rsa'
return key_path
def File_Path():
'''
用于crontab views中 crontab_get调用
:return: 用于存放收集到的计划任务,避免每次访问都重新获取主机计划任务
'''
file_path = "/data/filename.txt"
return file_path
host_cron/utils/select_crontab.py
import paramiko
import logging
def get_remote_crontab_tasks(hostname_list, username, private_key_path):
'''
用于获取主机上的crontab计划任务
:param hostname_list: 主机列表['1.1.1.1', '2.2.2.2']
:param username: 主机用户
:param private_key_path: 远程密钥
:return: 返回dict,{'hostname1': [crontab], 'hostname2': [crontab]}
'''
try:
# 创建SSH客户端对象
client = paramiko.SSHClient()
# 自动添加和保存远程服务器的SSH密钥
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 使用私钥文件进行身份验证
private_key = paramiko.RSAKey.from_private_key_file(private_key_path)
crontab_dict = {}
for hostname in hostname_list:
try:
# 连接远程服务器
client.connect(hostname, username=username, pkey=private_key, timeout=5)
# 执行命令获取crontab任务
stdin, stdout, stderr = client.exec_command('crontab -l')
# 获取命令输出结果
crontab_tasks = stdout.read().decode().splitlines()
crontab_dict[hostname] = crontab_tasks
# 关闭SSH连接
client.close()
except:
crontab_dict[hostname] = ['err: 登陆失败']
return crontab_dict
host_cron/views.py
from host_cron.utils.select_crotnab import get_remote_crontab_tasks
from crontab.config import Host_Add, Host_User, Host_Key_Path, File_Path
from django.shortcuts import render
import json
import os
def crontab_select(request):
'''
首页,用于直接从计划任务存储文件中读取计划任务,如果文件不存在则创建,并传入空字典,写入时需要转为json,读取时需要转为字典
:param request: 读取文件
:return: 计划任务字典到html
'''
file_path = File_Path()
if os.path.exists(file_path):
# 从文件中读取字符串内容
with open(file_path, "r") as file:
str_data = file.read()
# 将字符串解析为字典类型
crontab_tasks_dict = json.loads(str_data)
return render(request, 'crontab.html', {'crontab_tasks_dict': crontab_tasks_dict})
else:
crontab_tasks_dict = {}
# 打开写入文件
file = open(File_Path(), "w")
file.write(json.dumps(crontab_tasks_dict))
file.close()
return render(request, 'crontab.html', {'crontab_tasks_dict': crontab_tasks_dict})
def crontab_get(request):
'''
获取主机上的计划任务,并以json格式写入到文件中
:param request:
:return:
'''
# 远程服务器的连接参数
hostname_list = Host_Add() # 替换为实际的远程服务器主机名或IP地址
username = Host_User() # 替换为实际的远程服务器用户名
private_key_path = Host_Key_Path() # 替换为实际的私钥文件路径
# 获取远程服务器上的crontab任务
crontab_tasks_dict = get_remote_crontab_tasks(hostname_list, username, private_key_path)
# 打开文件
file = open(File_Path(), "w")
# 写入内容
file.write(json.dumps(crontab_tasks_dict))
# 关闭文件
file.close()
return render(request, 'crontab.html', {'crontab_tasks_dict': crontab_tasks_dict})
templates/crontab.html
html中使用js,点击按钮但不跳转链接,执行完成后自动请求刷新当前页面。
缺点:点击‘重新获取’按钮等待返回时没有获取中...的提示,懒了没搞。
<!DOCTYPE html>
<html>
<head>
<title>Crontab Tasks</title>
<style>
table {
width: 100%;
border-collapse: collapse;
}
th, td {
padding: 8px;
text-align: left;
border-bottom: 1px solid #ddd;
}
th {
background-color: #f2f2f2;
}
ul {
list-style-type: none;
padding: 0;
margin: 0;
}
ul li {
margin-left: 20px;
}
.custom-button {
background-color: blue;
color: white;
padding: 10px 20px;
margin-top: 10px;
margin-bottom: 10px;
border: none;
cursor: pointer;
}
</style>
<script>
function triggerURL() {
fetch('/crontab_get/')
.then(response => {
if (response.ok) {
console.log('URL triggered successfully');
// 执行其他操作,根据需要进行处理
location.reload(); // 重新加载当前页面
} else {
console.error('URL triggering failed');
}
})
.catch(error => {
console.error('Error:', error);
});
}
</script>
</head>
<body>
<button class="custom-button" onclick="triggerURL()">重新获取</button>
<table>
<thead>
<tr>
<th>IP Address</th>
<th>Crontab Tasks</th>
</tr>
</thead>
<tbody>
{% for ip, tasks in crontab_tasks_dict.items %}
<tr>
<td>{{ ip }}</td>
<td>
<ul>
{% for task in tasks %}
<li>{{ task }}</li>
{% endfor %}
</ul>
</td>
</tr>
{% empty %}
<tr>
<td colspan="2">No data available</td>
</tr>
{% endfor %}
</tbody>
</table>
</body>
</html>
docker部署
将代码目录压缩成crontab.tar.gz
Dockerfile
FROM python:3.9.13
RUN mkdir /data
RUN pip3 install Django paramiko -i https://mirrors.aliyun.com/pypi/simple/
ADD crontab.tar.gz /data
CMD python3 /data/crontab/manage.py runserver 0.0.0.0:18888
将代码中的config.py提取出来,根据你的docker启动命令定义和修改
构建镜像
docker build -t crontab:v1 .
启动
docker run -d \
--name crontab \
--network=host \
-v /data/crontab_py/id_rsa:/data/id_rsa \
-v /data/crontab_py/config.py:/data/crontab/crontab/config.py \
crontab:v1
网络直接用host宿主机网络,使用容器网络ip因为是陌生ip,会触发云上服务器的异常登陆报警。
浏览器 ip:18888 访问即可
搞定!