在计算机领域中，Ansible 是一款非常实用的自动化运维工具。今天咱们就来聊聊怎么开发 Ansible 回调插件，把自定义执行结果收集起来，还能在必要的时候发送告警。
一、什么是 Ansible 回调插件
Ansible 回调插件就像是一个小助手，它能在 Ansible 执行任务的过程中，对各种事件做出响应。比如说，当一个任务执行完成后，回调插件就能获取到执行结果，然后做一些额外的处理，像记录日志、发送通知啥的。举个例子，就好比你让一个朋友帮你做件事，事情做完了，朋友会给你反馈结果，这个反馈的过程就有点像回调插件的作用。
二、为什么要开发自定义回调插件
应用场景
在实际的运维工作中，我们可能有各种各样的需求。比如说，我们希望在 Ansible 执行完任务后，把结果存储到特定的数据库里，方便后续分析；或者当任务执行失败时，及时发送告警邮件或消息，让运维人员能第一时间知道出问题了。再比如，我们想对执行结果进行一些自定义的统计，像统计成功和失败的任务数量，这些都可以通过自定义回调插件来实现。
技术优缺点
优点：
- 灵活性高：可以根据自己的需求定制功能，满足各种特殊的业务场景。
- 扩展性强：可以很方便地添加新的功能，比如增加新的告警方式。
- 提高效率：自动化处理执行结果和告警，减少人工干预。
缺点：
- 开发成本相对较高：需要一定的编程基础，对于新手来说可能有一定难度。
- 维护复杂：如果插件功能复杂，后续的维护和更新可能会比较麻烦。
三、开发自定义回调插件的步骤
1. 创建插件文件
首先，我们要创建一个 Python 文件，作为我们的回调插件。这个文件要放在 Ansible 的回调插件目录下，一般是 /usr/share/ansible/plugins/callback/ 。下面是一个简单的示例（Python 技术栈）：
# 导入必要的模块
from ansible.plugins.callback import CallbackBase
class MyCallback(CallbackBase):
"""自定义回调插件类"""
def __init__(self):
# 调用父类的初始化方法
super(MyCallback, self).__init__()
def v2_runner_on_ok(self, result):
"""当任务执行成功时调用"""
host = result._host.get_name()
task = result._task.get_name()
print(f"任务 {task} 在主机 {host} 上执行成功")
def v2_runner_on_failed(self, result, ignore_errors=False):
"""当任务执行失败时调用"""
host = result._host.get_name()
task = result._task.get_name()
print(f"任务 {task} 在主机 {host} 上执行失败")
2. 配置 Ansible 使用插件
要让 Ansible 使用我们的插件，需要在 Ansible 的配置文件 ansible.cfg 中进行配置。在文件中添加以下内容：
[defaults]
callback_whitelist = mycallback
这里的 mycallback 是我们插件文件的名称（不包含 .py 后缀）。
3. 测试插件
编写一个简单的 Ansible 剧本，然后执行它，看看我们的插件是否能正常工作。以下是一个简单的 Ansible 剧本示例：
---
- name: 测试回调插件
hosts: all
tasks:
- name: 执行命令
command: ls /tmp
执行这个剧本：
ansible-playbook test.yml
如果一切正常，你应该能看到插件输出的信息。
四、实现自定义执行结果收集
存储结果到文件
我们可以把执行结果存储到文件中，方便后续查看。修改之前的回调插件代码：
# 导入必要的模块
from ansible.plugins.callback import CallbackBase
import json
class MyCallback(CallbackBase):
"""自定义回调插件类"""
def __init__(self):
# 调用父类的初始化方法
super(MyCallback, self).__init__()
self.results = []
def v2_runner_on_ok(self, result):
"""当任务执行成功时调用"""
host = result._host.get_name()
task = result._task.get_name()
output = result._result
data = {
"host": host,
"task": task,
"status": "success",
"output": output
}
self.results.append(data)
def v2_playbook_on_stats(self, stats):
"""当剧本执行完成时调用"""
with open('results.json', 'w') as f:
json.dump(self.results, f, indent=4)
存储结果到数据库
如果需要更方便地查询和分析结果，我们可以把结果存储到数据库中。这里以 SQLite 为例：
# 导入必要的模块
from ansible.plugins.callback import CallbackBase
import sqlite3
class MyCallback(CallbackBase):
"""自定义回调插件类"""
def __init__(self):
# 调用父类的初始化方法
super(MyCallback, self).__init__()
self.conn = sqlite3.connect('results.db')
self.cursor = self.conn.cursor()
# 创建表
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS results (
host TEXT,
task TEXT,
status TEXT,
output TEXT
)
''')
def v2_runner_on_ok(self, result):
"""当任务执行成功时调用"""
host = result._host.get_name()
task = result._task.get_name()
output = str(result._result)
self.cursor.execute('''
INSERT INTO results (host, task, status, output)
VALUES (?,?,?,?)
''', (host, task, 'success', output))
self.conn.commit()
def __del__(self):
"""析构函数，关闭数据库连接"""
self.conn.close()
五、实现告警功能
发送邮件告警
当任务执行失败时，我们可以发送邮件告警。这里使用 Python 的 smtplib 模块：
# 导入必要的模块
from ansible.plugins.callback import CallbackBase
import smtplib
from email.mime.text import MIMEText
class MyCallback(CallbackBase):
"""自定义回调插件类"""
def __init__(self):
# 调用父类的初始化方法
super(MyCallback, self).__init__()
self.sender = 'your_email@example.com'
self.receiver = 'recipient_email@example.com'
self.password = 'your_email_password'
def v2_runner_on_failed(self, result, ignore_errors=False):
"""当任务执行失败时调用"""
host = result._host.get_name()
task = result._task.get_name()
output = str(result._result)
msg = MIMEText(f"任务 {task} 在主机 {host} 上执行失败，输出信息：{output}")
msg['Subject'] = 'Ansible 任务执行失败告警'
msg['From'] = self.sender
msg['To'] = self.receiver
try:
server = smtplib.SMTP('smtp.example.com', 587)
server.starttls()
server.login(self.sender, self.password)
server.sendmail(self.sender, self.receiver, msg.as_string())
server.quit()
print("告警邮件发送成功")
except Exception as e:
print(f"告警邮件发送失败：{e}")
发送消息告警
除了邮件告警，我们还可以使用第三方消息平台，比如 Slack 或钉钉。以下是一个使用钉钉机器人发送消息告警的示例：
# 导入必要的模块
from ansible.plugins.callback import CallbackBase
import requests
class MyCallback(CallbackBase):
"""自定义回调插件类"""
def __init__(self):
# 调用父类的初始化方法
super(MyCallback, self).__init__()
self.webhook_url = 'https://oapi.dingtalk.com/robot/send?access_token=your_access_token'
def v2_runner_on_failed(self, result, ignore_errors=False):
"""当任务执行失败时调用"""
host = result._host.get_name()
task = result._task.get_name()
output = str(result._result)
data = {
"msgtype": "text",
"text": {
"content": f"任务 {task} 在主机 {host} 上执行失败，输出信息：{output}"
}
}
headers = {'Content-Type': 'application/json'}
try:
response = requests.post(self.webhook_url, json=data, headers=headers)
if response.status_code == 200:
print("钉钉消息发送成功")
else:
print(f"钉钉消息发送失败：{response.text}")
except Exception as e:
print(f"钉钉消息发送失败：{e}")
六、注意事项
- 插件的兼容性：确保插件与你使用的 Ansible 版本兼容，不同版本的 Ansible 可能对回调插件的接口有不同的要求。
- 错误处理：在插件中要做好错误处理，避免因为一些异常情况导致插件崩溃。
- 性能问题：如果插件处理的数据量较大，要注意性能优化，避免影响 Ansible 的执行效率。
七、文章总结
通过开发 Ansible 回调插件，我们可以实现自定义的执行结果收集和告警功能。这不仅能提高运维工作的效率，还能让我们更好地掌握 Ansible 任务的执行情况。在开发过程中，我们要根据实际需求选择合适的存储方式和告警方式，同时注意插件的兼容性、错误处理和性能问题。希望这篇文章能帮助你开发出实用的 Ansible 回调插件。
评论