Python结合Pexpect实现多服务器进程状态巡检

通过Python与pexpect实现跨节点进程状态巡检

在日常开发与系统维护中，经常需要在跳板机（堡垒机）上核对多台后端服务器的关键服务是否正常运行。当安全策略禁止配置SSH密钥互信，且当前用户仅拥有普通权限无法直接修改系统Hosts文件时，传统的批量命令会因认证交互和别名解析失败。本文将演示如何利用Python脚本配合pexpect终端交互库与PyYAML配置文件，自动处理SSH登录提示、下发ps命令统计进程数，并以终端彩色日志直观反馈各服务器服务状态。

准备条件与环境配置

第1步：搭建本地依赖运行环境

脚本基于Python标准库扩展了pexpect（用于模拟终端输入输出）与PyYAML（用于解析配置）。为便于多人共享且不污染个人目录，可将依赖包解压至公共目录或虚拟环境中。若依赖路径非默认系统路径，需在代码头部动态追加环境变量：

# 将依赖包所在目录加入Python搜索路径，按需调整实际解压位置
import sys
sys.path.insert(0, '/path/to/your/dependences') # 请替换为实际的依赖存放路径
import yaml
import pexpect

第2步：编写YAML服务端点清单

创建config.yaml文件，集中管理目标主机的逻辑别名、展示IP以及期望运行的服务与预期进程数。使用逻辑别名可绕过缺少/etc/hosts写入权限的限制，前提是在客户端本地~/.ssh/config或通过-F参数加载自定义映射：

bj-server-node:
  host: node_alias    # 需与sshconfig中的Host别名一致
  ip: 10.0.0.25       # 仅作显示用途，便于控制台核对
  service:            # 列表形式定义待检服务及预期进程数
    - nginx: 5
    - uwsgi: 25
    - mysql: 3
    - celery: 12
    - redis: 1

若仅需确认服务存活而不关心具体数量，可省略数值或单独标记：

• 带数字格式（如 nginx: 5）表示必须严格等于该进程数才算正常。
• 纯服务名格式（如 - redis）表示只要存在至少一个进程实例即判定为正常。

核心巡检脚本实现与逻辑拆解

以下为完整的Python巡检脚本。已优化原版缩进与闭包变量作用域问题，确保终端颜色打印准确，并兼容常见的交互提示符匹配逻辑。

#coding=utf-8
import sys
# 确保能导入本地的 pexpect 和 yaml 包
sys.path.append('/tmp/local/lib/site-packages') 

import yaml
import pexpect

# 加载配置文件
with open('/tmp/config.yaml', 'r') as f:
    dataDict = yaml.safe_load(f)

def myprint(color_code, msg):
    '''使用ANSI控制码终端显示彩色'''
    d = dict(r=31, g=32, gb=36, y=33, b=34, p=35, o=37)
    code = "\x1B[1;%dm" % d.get(color_code, 37)
    print("{}{}\x1B[0m".format(code, msg))

def get_encrypted_password(host_name):
    '''示例占位函数：根据主机名返回解密后的密码'''
    return "Your_Secure_Password"

def main():
    color_cycle = ['g', 'b', 'y', 'gb', 'p']
    light_idx = 0
    
    for server_key in dataDict:
        if not server_key.startswith('bj-'):
            continue
            
        color = color_cycle[light_idx % len(color_cycle)]
        server_info = dataDict[server_key]
        
        # 通过 -F 参数指定本地sshconfig文件，绕过系统hosts限制
        ssh_cmd = 'ssh -F /tmp/sshconfig {user}@{host}'.format(
            user='dongwm', 
            host=server_info['host']
        )
        
        child = pexpect.spawn(ssh_cmd)
        # 捕获密码提示或首次连接指纹确认提示
        res_index = child.expect(['Password: ', 'password: ', 'continue connecting (yes/no)?'])
        
        # 处理首次连接需手动确认指纹的情况
        if res_index == 2:
            child.sendline('yes')
            child.expect('password:')
            
        # 发送登录密码
        child.sendline(get_encrypted_password(server_info['host']))
        
        # 等待Shell就绪提示符（以 ~ 结尾通常是普通用户prompt特征）
        child.expect('~$?')
        
        # 遍历配置的服务列表进行进程数核对
        for svc_item in server_info['service']:
            is_exact_count = True
            target_proc_count = 0
            svc_name = str(svc_item).split(':')[0] if isinstance(svc_item, dict) else svc_item
            
            if isinstance(svc_item, dict):
                svc_name = list(svc_item.keys())[0]
                target_proc_count = svc_item[svc_name]
            else:
                is_exact_count = False
                target_proc_count = -1
            
            # 执行查询命令：过滤掉grep自身，统计行数即为进程数
            query_cmd = 'ps -ef | grep {name} | grep -v grep | wc -l'.format(name=svc_name)
            child.sendline(query_cmd)
            child.expect('~$?')
            
            # 读取wc -l输出的数字行
            raw_line = child.before.decode().strip()
            actual_num = int(raw_line) if raw_line.isdigit() else 0
            
            # 判定逻辑：精确计数匹配，或非精确计数且进程存在(>0)
            status_ok = (actual_num == target_proc_count) or (not is_exact_count and actual_num > 0)
            
            light_idx += 1
            if status_ok:
                myprint(color, "[{k}] [{ip}] [{svc}] [ok]".format(
                    k=server_key.center(12), ip=server_info.get('ip','').center(14), svc=svc_name.center(20)))
            else:
                display_num = actual_num if target_proc_count > 0 else '存在'
                myprint('r', "[{k}] [{ip}] [{svc}] [fail] ({n}!={exp})".format(
                    k=server_key.center(12), ip=server_info.get('ip','').center(14),
                    svc=svc_name.center(20), n=display_num, exp=target_proc_count))
                    
            child.sendline('exit')
            try: child.close(force=True)
            except: pass

if __name__ == '__main__':
    main()

运行操作步骤与完成判断

第3步：授权脚本并执行巡检

将上述代码保存为checklist.py，配置文件存为config.yaml。在跳板机终端中赋予执行权限并运行：

chmod +x checklist.py
python3 checklist.py   # 若环境为Python 2.6可改用 python checklist.py

第4步：核对终端彩色输出结果

脚本按顺序轮询列出所有匹配的服务器节点。绿色（g）/蓝色（b）等字样代表该服务器整体状态正常，红色（r）标识进程数不匹配。控制台输出格式固定为：[ServerKey] [DisplayIP] [ServiceName] [Status]。当目标节点全部显示[ok]且无红色警告时，即可判定巡检任务完成。

排错指南与安全注意事项

• 提示符匹配失效：不同系统的Shell提示符结尾可能不同。若脚本报错卡在child.expect('~')，请将其改为目标服务器实际返回的用户提示符（如$、#或\]），或使用child.prompt。
• 权限与网络限制：本方案专为普通用户+无Hosts权限+无SSH互信环境设计。若目标服务器开启了非常规端口或防火墙拦截SSH，需提前联系网络管理员放行。
• Python版本差异：原文基于Python 2.6编写，现代Linux环境普遍预装Python 3。脚本中的字符串拼接、字典迭代与child.before编码方式已做向上兼容处理，建议使用Python 3.6及以上版本运行。
• 账号安全：硬编码密码极不安全。生产环境中请务必引入环境变量注入、密钥管理服务（Vault）或免密证书通道替代明文传递。

以上就是利用Python结合pexpect库实现多服务器进程状态自动巡检的详细内容，更多关于自动化运维与跳板机安全核查的资料请关注本站其它相关文章！