Python运维实战

#!/usr/bin/env python3"""批量服务器磁盘检查工具功能：并发SSH到多台服务器，执行df -h命令，收集磁盘使用情况"""import paramikoimport threadingfrom concurrent.futures import ThreadPoolExecutor, as_completedimport json# 服务器列表（实际项目中从CMDB或配置文件读取）SERVERS = [    {"hostname": "192.168.1.101", "port": 22, "username": "root"},    {"hostname": "192.168.1.102", "port": 22, "username": "root"},    {"hostname": "192.168.1.103", "port": 22, "username": "root"},    # ... 更多服务器]# SSH密钥路径（推荐使用密钥认证）KEY_PATH = "/root/.ssh/id_rsa"def ssh_execute_command(hostname, port, username, command):    """    通过SSH执行远程命令    :param hostname: 服务器IP或域名    :param port: SSH端口    :param username: 用户名    :param command: 要执行的命令    :return: (success, output) 元组    """    try:        ssh = paramiko.SSHClient()        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())        # 使用密钥认证（推荐）        ssh.connect(            hostname=hostname,            port=port,            username=username,            key_filename=KEY_PATH,            timeout=10        )        # 执行命令        stdin, stdout, stderr = ssh.exec_command(command, timeout=30)        output = stdout.read().decode('utf-8')        error = stderr.read().decode('utf-8')        ssh.close()        if error:            return False, f"ERROR: {error}"        return True, output    except Exception as e:        return False, f"Exception: {str(e)}"def check_disk_usage(server):    """检查单台服务器的磁盘使用情况"""    hostname = server["hostname"]    port = server["port"]    username = server["username"]    success, output = ssh_execute_command(hostname, port, username, "df -h")    return {        "hostname": hostname,        "success": success,        "output": output    }def batch_check_disks(servers, max_workers=10):    """    批量检查磁盘使用情况（并发执行）    :param servers: 服务器列表    :param max_workers: 最大线程数    """    results = []    with ThreadPoolExecutor(max_workers=max_workers) as executor:        # 提交所有任务        future_to_server = {            for server in servers        }        # 收集结果        for future in as_completed(future_to_server):            result = future.result()            results.append(result)            # 实时输出进度            if result["success"]:                print(f"✓ {result['hostname']} - Success")            else:                print(f"✗ {result['hostname']} - Failed: {result['output']}")    return resultsif __name__ == "__main__":    print("=" * 60)    print("批量磁盘检查工具")    print("=" * 60)    results = batch_check_disks(SERVERS, max_workers=20)    print("\n" + "=" * 60)    print("检查完成！结果汇总：")    print("=" * 60)    # 保存结果到JSON文件    with open("disk_check_results.json", "w") as f:        json.dump(results, f, indent=2, ensure_ascii=False)    print(f"结果已保存到 disk_check_results.json")

#!/usr/bin/env python3"""阿里云ECS管理工具 - 通过API批量操作"""import requestsimport jsonimport hmacimport hashlibfrom datetime import datetimeimport urllib.parse# 阿里云API配置（实际项目中从环境变量或配置文件读取）ACCESS_KEY_ID = "YOUR_ACCESS_KEY_ID"ACCESS_KEY_SECRET = "YOUR_ACCESS_KEY_SECRET"REGION_ID = "cn-hangzhou"class AliyunECSManager:    """阿里云ECS管理类"""    def __init__(self, access_key_id, access_key_secret, region_id):        self.access_key_id = access_key_id        self.access_key_secret = access_key_secret        self.region_id = region_id        self.base_url = "https://ecs.aliyuncs.com"    def _sign_request(self, params):        """生成API请求签名（阿里云RPC签名机制）"""        # 省略签名逻辑（实际项目中建议使用aliyun-python-sdk-core库）        pass    def describe_instances(self, page_number=1, page_size=10):        """查询ECS实例列表"""        params = {            "Action": "DescribeInstances",            "Version": "2014-05-26",            "RegionId": self.region_id,            "PageNumber": page_number,            "PageSize": page_size        }        # 添加签名        self._sign_request(params)        try:            response = requests.get(self.base_url,                params=params,                timeout=10            )            response.raise_for_status()            return response.json()        except requests.exceptions.RequestException as e:            print(f"API请求失败: {e}")            return None    def start_instance(self, instance_id):        """启动ECS实例"""        params = {            "Action": "StartInstance",            "Version": "2014-05-26",            "RegionId": self.region_id,            "InstanceId": instance_id        }        self._sign_request(params)        try:            response = requests.get(self.base_url, params=params, timeout=10)            response.raise_for_status()            result = response.json()            print(f"✓ 实例 {instance_id} 启动请求已提交，RequestId: {result.get('RequestId')}")            return True        except requests.exceptions.RequestException as e:            print(f"✗ 启动实例 {instance_id} 失败: {e}")            return Falseif __name__ == "__main__":    # 初始化ECS管理器    ecs_manager = AliyunECSManager(ACCESS_KEY_ID, ACCESS_KEY_SECRET, REGION_ID)    # 查询实例列表    print("查询ECS实例列表...")    result = ecs_manager.describe_instances()    if result and "Instances" in result:        instances = result["Instances"]["Instance"]        print(f"\n找到 {len(instances)} 个ECS实例：\n")        for instance in instances:            print(f"实例ID: {instance['InstanceId']}")            print(f"实例名称: {instance['InstanceName']}")            print(f"状态: {instance['Status']}")            print("-" * 40)            # 如果实例是已停止状态，则启动它            if instance["Status"] == "Stopped":                ecs_manager.start_instance(instance["InstanceId"])    else:        print("查询失败或没有找到实例")

#!/usr/bin/env python3"""MySQL慢查询分析工具"""import pymysqlimport pandas as pdfrom datetime import datetime, timedeltaimport json# 数据库配置（实际项目中从配置文件读取）DB_CONFIGS = [    {        "host": "192.168.1.201",        "port": 3306,        "user": "monitor",        "password": "monitor_pass",        "database": "mysql"    },    {        "host": "192.168.1.202",        "port": 3306,        "user": "monitor",        "password": "monitor_pass",        "database": "mysql"    }]def get_slow_queries(db_config, start_time, end_time, limit=10):    """    从MySQL慢查询日志表（需要开启performance_schema）获取慢查询    如果未开启performance_schema，可以解析slow_query_log文件    """    try:        # 建立数据库连接        connection = pymysql.connect(            host=db_config["host"],            port=db_config["port"],            user=db_config["user"],            password=db_config["password"],            database=db_config["database"],            charset='utf8mb4',            cursorclass=pymysql.cursors.DictCursor # 返回字典格式        )        with connection.cursor() as cursor:            # 查询慢查询（假设已经开启了慢查询日志并解析到表中）            # 实际项目中可以查询performance_schema.events_statements_history_long            sql = """                SELECT                    DIGEST_TEXT AS query_pattern,                    COUNT_STAR AS exec_count,                    AVG_TIMER_WAIT/1000000000 AS avg_latency_sec,                    MAX_TIMER_WAIT/1000000000 AS max_latency_sec,                    SUM_ROWS_SENT AS rows_sent                FROM performance_schema.events_statements_summary_by_digest                WHERE AVG_TIMER_WAIT/1000000000 > 1 -- 平均执行时间超过1秒                ORDER BY AVG_TIMER_WAIT DESC                LIMIT %s            """            cursor.execute(sql, (limit,))            slow_queries = cursor.fetchall()            return slow_queries    except pymysql.MySQLError as e:        print(f"数据库查询失败 ({db_config['host']}): {e}")        return []    finally:        if 'connection' in locals():            connection.close()def analyze_slow_queries(db_configs):    """分析多个数据库的慢查询，生成汇总报告"""    all_slow_queries = []    for db_config in db_configs:        host = db_config["host"]        print(f"\n正在分析数据库: {host}...")        slow_queries = get_slow_queries(db_config, None, None, limit=10)        for query in slow_queries:            query["host"] = host            all_slow_queries.append(query)        print(f"✓ 找到 {len(slow_queries)} 条慢查询")    # 转换为DataFrame进行分析    df = pd.DataFrame(all_slow_queries)    if df.empty:        print("\n未发现慢查询")        return None    # 按平均延迟排序    df = df.sort_values("avg_latency_sec", ascending=False)    # 生成报告    report = {        "generated_at": datetime.now().isoformat(),        "total_slow_queries": len(df),        "top_10_slow_queries": df.head(10).to_dict("records")    }    return reportif __name__ == "__main__":    print("=" * 70)    print("MySQL慢查询分析工具")    print("=" * 70)    report = analyze_slow_queries(DB_CONFIGS)    if report:        # 保存报告        report_file = f"slow_query_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"        with open(report_file, "w") as f:            json.dump(report, f, indent=2, ensure_ascii=False)        print(f"\n报告已保存: {report_file}")        print("\nTOP 10 慢查询：")        print("-" * 70)        for idx, query in enumerate(report["top_10_slow_queries"], 1):            print(f"\n#{idx}")            print(f" 数据库: {query['host']}")            print(f" 查询模式: {query['query_pattern'][:80]}...")            print(f" 执行次数: {query['exec_count']}")            print(f" 平均延迟: {query['avg_latency_sec']:.2f}秒")            print(f" 最大延迟: {query['max_latency_sec']:.2f}秒")