当前位置：首页 > news >正文

python自动化运维通过paramiko库和time库实现服务器自动化管理

news 2025/8/25 15:57:38

一.前言

二. 代码实现以及解析

2.1导入必要的库

2.2定义服务器信息

2.3创建 SSH 客户端连接函数

2.4执行远程命令函数

2.5获取系统信息函数

2.6重启服务函数

2.7 主函数

三.致谢

一.前言

在数字化时代，IT 基础设施的规模和复杂性不断增长，传统的手动管理方法已不再适用。自动化运维，作为 IT 运维管理的新范式，正变得越来越重要。它不仅能够显著提升运维效率，降低人为错误，还能确保业务连续性和系统的高可用性。

Python，以其简洁的语法和强大的功能，成为自动化运维的优选语言。借助 Python，我们可以快速开发出灵活且强大的自动化脚本，以应对各种运维场景。

在本文中，我们将探讨如何使用 Python 的 paramiko 库来实现 SSH 连接和远程命令执行，以及如何利用 time 库来处理时间相关操作，从而实现服务器的自动化管理。这些库的结合使用将使我们能够编写出功能丰富、健壮且易于维护的自动化脚本。

二. 代码实现以及解析

import paramiko
import time# 定义服务器信息
hostname = 'your_server_hostname_or_ip'
port = 22
username = 'your_username'
password = 'your_password'# 创建 SSH 客户端连接
def connect_to_server():ssh_client = paramiko.SSHClient()ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())try:ssh_client.connect(hostname, port, username, password)print(f"Connected to {hostname} successfully.")return ssh_clientexcept Exception as e:print(f"Error connecting to {hostname}: {str(e)}")return None# 执行远程命令
def run_command(ssh_client, command):stdin, stdout, stderr = ssh_client.exec_command(command)output = stdout.read().decode('utf-8')error = stderr.read().decode('utf-8')if error:print(f"Error executing command '{command}': {error.strip()}")else:print(f"Command '{command}' executed successfully. Output:\n{output.strip()}")# 获取系统信息示例
def get_system_info(ssh_client):run_command(ssh_client, 'uname -a')run_command(ssh_client, 'df -h')run_command(ssh_client, 'free -m')# 重启服务示例
def restart_service(ssh_client, service_name):run_command(ssh_client, f'sudo systemctl restart {service_name}')time.sleep(5)  # 等待一段时间以确保服务重启完成run_command(ssh_client, f'sudo systemctl status {service_name}')# 主函数
def main():ssh_client = connect_to_server()if ssh_client:get_system_info(ssh_client)restart_service(ssh_client, 'apache2')  # 以 Apache2 为例，可以根据需要替换成你的服务名ssh_client.close()print("Script execution completed.")else:print("Exiting script due to connection error.")if __name__ == "__main__":main()

2.1导入必要的库

import paramiko
import time

paramiko 是一个用于 SSH2 协议的 Python 库，用于远程操作服务器。
time 是 Python 标准库，用于在执行操作之间添加延迟，如等待服务重启完成。

2.2定义服务器信息

hostname = 'your_server_hostname_or_ip'
port = 22
username = 'your_username'
password = 'your_password'

这些变量包括远程服务器的主机名（或IP地址）、SSH端口号、登录用户名和密码。实际应用中，密码应该通过安全的方式进行管理，如使用环境变量或密钥认证。

2.3创建 SSH 客户端连接函数

def connect_to_server():ssh_client = paramiko.SSHClient()ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())try:ssh_client.connect(hostname, port, username, password)print(f"Connected to {hostname} successfully.")return ssh_clientexcept Exception as e:print(f"Error connecting to {hostname}: {str(e)}")return None

connect_to_server() 函数通过 paramiko.SSHClient() 创建一个 SSH 客户端对象。
set_missing_host_key_policy(paramiko.AutoAddPolicy()) 设置自动添加新主机密钥的策略，适用于首次连接服务器时。
ssh_client.connect() 方法尝试连接远程服务器，并打印连接成功或失败的信息。

2.4执行远程命令函数

def run_command(ssh_client, command):stdin, stdout, stderr = ssh_client.exec_command(command)output = stdout.read().decode('utf-8')error = stderr.read().decode('utf-8')if error:print(f"Error executing command '{command}': {error.strip()}")else:print(f"Command '{command}' executed successfully. Output:\n{output.strip()}")

run_command() 函数接收一个 SSH 客户端对象 ssh_client 和要执行的命令 command。
使用 exec_command() 方法在远程服务器上执行命令，并获取标准输入、输出和错误流。
输出结果进行解码（通常为UTF-8），并根据执行情况打印成功或失败的信息。

2.5获取系统信息函数

def get_system_info(ssh_client):run_command(ssh_client, 'uname -a')run_command(ssh_client, 'df -h')run_command(ssh_client, 'free -m')

get_system_info() 函数调用 run_command() 函数来获取系统信息：
- uname -a：获取操作系统的详细信息。
- df -h：获取磁盘空间使用情况。
- free -m：获取内存使用情况。

2.6重启服务函数

def restart_service(ssh_client, service_name):run_command(ssh_client, f'sudo systemctl restart {service_name}')time.sleep(5)  # 等待一段时间以确保服务重启完成run_command(ssh_client, f'sudo systemctl status {service_name}')

restart_service() 函数接收一个服务名 service_name，使用 sudo systemctl 命令重启指定的服务，并检查服务状态以确认是否重启成功。

2.7 主函数

def main():ssh_client = connect_to_server()if ssh_client:get_system_info(ssh_client)restart_service(ssh_client, 'apache2')  # 以 Apache2 为例，可以根据需要替换成你的服务名ssh_client.close()print("Script execution completed.")else:print("Exiting script due to connection error.")if __name__ == "__main__":main()