Ansible自定义模块需用Python实现符合规范的脚本，通过AnsibleModule接收name等参数，调用subprocess/socket/urllib检查进程、端口或HTTP状态，并以JSON返回changed、failed、msg及status、pid等字段。
编写 Ansible 自定义模块来实现服务状态监控检查，核心是用 Python 实现一个可被 Ansible 调用的、符合其模块接口规范的脚本，能返回标准 JSON 格式的执行结果（包括 changed、failed、msg 和自定义字段如 status、pid 等）。
明确模块职责与输入参数
先想清楚这个模块要做什么：比如检查某个进程是否在运行（通过 ps 或 systemctl is-active）、端口是否监听、HTTP 接口是否返回 200。常见输入参数包括：
- name：服务名或进程名（必填）
-
state：期望状态，如
running、listening、healthy（可选，默认running） - port：若检查端口，指定端口号（可选）
- url：若做 HTTP 健康检查，传入 URL（可选）
- timeout：超时时间（秒，默认 5）
编写符合 Ansible 规范的 Python 模块
Ansible 模块本质是一个独立的 Python 脚本，需满足以下要求：
- 以
#!/usr/bin/env python开头（shebang） - 导入
ansible.module_utils.basic并调用AnsibleModule - 定义
argument_spec明确参数类型和必需性 - 执行逻辑中捕获异常，避免崩溃；所有输出必须通过
module.exit_json()或module.fail_json()返回 - 不依赖外部 pip 包（除非目标环境已预装），优先用标准库（
subprocess、socket、urllib.request等）
示例片段（检查进程是否存在）：
（模块路径：library/check_service.py）#!/usr/bin/env python
from ansible.module_utils.basic import AnsibleModule
import subprocess
import os
<p>def check_process(name):
try:
out = subprocess.check_output(['pgrep', '-f', name], stderr=subprocess.STDOUT)
pids = out.strip().split()
return len(pids) > 0, pids[0].decode() if pids else None
except subprocess.CalledProcessError:
return False, None</p><p>def main():
module = AnsibleModule(
argument_spec=dict(
name=dict(required=True, type='str'),
state=dict(default='running', choices=['running', 'stopped']),
),
supports_check_mode=False
)</p><pre class="brush:php;toolbar:false;">name = module.params['name']
expected = module.params['state']
is_running, pid = check_process(name)
changed = False
msg = f"Service '{name}' is {'running' if is_running else 'stopped'}"
if expected == 'running' and not is_running:
module.fail_json(msg=f"Expected '{name}' to be running, but it's not")
elif expected == 'stopped' and is_running:
module.fail_json(msg=f"Expected '{name}' to be stopped, but PID {pid} found")
module.exit_json(changed=changed, msg=msg, status='running' if is_running else 'stopped', pid=pid)if name == 'main': main()
在 playbook 中调用并处理结果
把模块放在 library/ 目录下（或配置 ANSIBLE_LIBRARY），即可像内置模块一样使用：
- name: Check nginx process status
check_service:
name: "nginx"
state: "running"
register: nginx_status
<ul><li><p>name: Fail if nginx isn't healthy
fail:
msg: "Nginx service check failed"
when: nginx_status.failed</p></li><li><p>name: Debug PID
debug:
var: nginx_status.pid
注意：返回字段（如 pid、status）可在后续任务中通过 register 变量引用，用于条件判断或日志记录。
增强健壮性与兼容性
生产环境建议补充以下细节：
- 对不同 Linux 发行版做适配（如用
systemctl is-active {{ name }}替代pgrep，需判断 systemd 是否存在） - 增加重试逻辑（如端口检查失败时等待后重试 2 次）
- 支持非 root 用户执行（例如检查用户级进程，避免硬编码 sudo）
- 添加日志输出（用
module.log()记录调试信息，但不暴露敏感内容） - 单元测试：用
pytest+mock模拟 subprocess 调用，验证各种返回路径
不复杂但容易忽略。关键不是写得多 fancy，而是返回结构清晰、错误有提示、参数有校验、行为可预测。
