事件源

事件源插件负责生成触发 ansible-rulebook 中规则评估的事件。
它们既可以接收事件，也可以与外部系统（如消息队列）交互，
从而为规则引擎生成事件数据。

为了帮助用户入门，ansible.eda 集合提供了一套事件源插件，
涵盖了 Ansible 事件驱动自动化的常见集成场景。您可以在此处探索可用的源插件：
https://galaxy.ansible.com/ui/repo/published/ansible/eda/content/

内置事件源

ansible-rulebook 提供了以下用于测试和开发的内置事件源：

eda.builtin.range
eda.builtin.generic
eda.builtin.webhook
eda.builtin.pg_listener

eda.builtin.range

生成带有递增索引 i 的事件。适用于测试和生成事件序列。

Name	描述	必填
limit	范围的上限（不包含）。事件的索引将从 0 到 limit-1	是
delay	事件之间的等待秒数。默认值：0	否

示例

sources:
  - name: range_source
    eda.builtin.range:
      limit: 5
      delay: 1

这将生成 5 个事件：{"i": 0}, {"i": 1}, {"i": 2}, {"i": 3}, {"i": 4}

eda.builtin.generic

用于插入自定义测试数据的通用源插件。适用于开发、测试和演示。

Name	描述	必填
payload	要插入队列的事件数组	是（除非使用了 payload_file）
payload_file	包含事件数组的 YAML 文件路径	是（除非使用了 payload）
loop_count	循环遍历 payload 的次数。默认值：1	否
randomize	随机排列事件顺序。默认值：false	否
timestamp	为每个事件添加时间戳。默认值：false	否
time_format	时间戳格式：“local”、“iso8601” 或 “epoch”。默认值：“local”	否
create_index	要添加到每个事件的索引字段名称（从 0 开始）	否
event_delay	事件之间的等待秒数。默认值：0	否

示例

sources:
  - name: test_data
    eda.builtin.generic:
      payload:
        - name: "Event 1"
          data: "test"
        - name: "Event 2"
          data: "example"
      loop_count: 2
      create_index: "seq"

eda.builtin.webhook

通过 HTTP webhook 接收事件。提供一个 HTTP 服务器端点，用于接收带有 JSON 有效负载的 POST 请求。

Name	描述	必填
host	监听的主机名。默认值：“0.0.0.0”	否
port	监听的 TCP 端口	是
token	可选的 Bearer 身份验证令牌	否
hmac_secret	用于有效负载验证的可选 HMAC 密钥	否
hmac_algo	HMAC 算法（sha256, sha512 等）。默认值：“sha256”	否
hmac_header	包含 HMAC 签名的 HTTP 头。默认值：“x-hub-signature-256”	否
hmac_format	HMAC 签名格式：“hex” 或 “base64”。默认值：“hex”	否
certfile	用于 TLS 支持的证书文件路径	否
keyfile	密钥文件路径（与 certfile 一起使用）	否
cafile	用于 mTLS 的 CA 证书文件路径	否

示例

sources:
  - name: webhook_events
    eda.builtin.webhook:
      host: 0.0.0.0
      port: 5000
      token: "my-secret-token"

带有 HMAC 验证的示例

sources:
  - name: github_webhook
    eda.builtin.webhook:
      host: 0.0.0.0
      port: 5000
      hmac_secret: "github-webhook-secret"
      hmac_algo: "sha256"
      hmac_header: "x-hub-signature-256"
      hmac_format: "hex"

注意

Webhook 事件源将接收到的有效负载放在事件数据的 payload 键下。如果您需要从有效负载中提取主机，请使用 eda.builtin.insert_hosts_to_meta 过滤器。

eda.builtin.pg_listener

PostgreSQL LISTEN/NOTIFY 事件源。监听来自 PostgreSQL 数据库的通知。

Name	描述	必填
dsn	PostgreSQL 连接字符串（例如：“host=localhost dbname=mydb user=myuser password=mypass”）	是
channels	要监听的 PostgreSQL 频道列表	是
delay	轮询延迟（秒）。默认值：0	否

示例

sources:
  - name: postgres_notifications
    eda.builtin.pg_listener:
      dsn: "host=localhost dbname=events user=eda password=secret"
      channels:
        - rulebook_events
        - alerts

注意

pg_listener 源要求安装 psycopg 库。

开发自定义事件源插件

您可以用 Python 构建自己的事件源插件。插件是一个单独的 Python 文件。

在决定是否构建专用插件时，请先考虑将数据源配置为发送数据到现已有通用插件支持的系统中。例如，如果您有一个能将数据发送到 Kafka 主题的系统，则可以使用 ansible.eda.kafka 插件来接收数据。有许多连接器可用于将系统与消息总线连接起来，这是利用现有插件的好方法。

在开始之前，让我们看看一些最佳实践和模式。

开发事件源插件有 3 种模式：

事件总线插件（推荐）
这些插件监听来自消息总线或队列的事件流，连接由插件本身建立。示例包括 ansible.eda.kafka 和 ansible.eda.aws_sqs_queue 插件。

这是推荐且最可靠的模式。 事件总线插件提供：
- 持久性：消息由消息总线持久化，直到被消费
- 可靠性：内置重试和错误处理机制
- 可扩展性：消息总线旨在处理高容量事件流
- 有序性：事件通常按顺序传递
- 确认机制：插件可以确认处理是否成功
数据的持久性和可靠性由事件源负责，数据的可用性可遵循事件源及其自身内部配置的模式。

如有可能，请考虑使用连接器或集成工具将平台的事件发送到受良好支持的消息总线（如 Kafka、Azure Service Bus 或 AWS SQS），而不是构建自定义源插件。这使您可以利用现有的、经过充分测试的插件。
回调插件（谨慎使用）
这些插件提供一个回调端点（通常是 Webhook 接收器），外部事件源可以在有数据可用时调用它。eda.builtin.webhook 插件就是这种模式的例子。

回调插件的重要注意事项：
- 数据丢失风险：如果事件发生时回调端点不可用，事件可能会丢失
- 网络要求：需要适当的入口策略、防火墙规则和网络可达性
- 无内置重试：如果外部系统未实现重试逻辑，事件可能无法重新投递
- 安全性担忧：暴露 HTTP 端点需要仔细的身份验证和授权
注意

建议：使用 Ansible Automation Platform 的集成 事件流 (Event Streams) 功能，而不是构建自定义回调插件。事件流提供了一个托管的 Webhook 基础设施，具有更好的可靠性和安全性。

更多信息，请参阅：https://docs.redhat.com/en/documentation/red_hat_ansible_automation_platform/2.6/html/using_automation_decisions/simplified-event-routing
抓取（Scraper）插件
这些插件连接到源并从中抓取数据，通常是在经过一段时间后。示例包括 ansible.eda.url_check 和 ansible.eda.file_watch 插件。

抓取插件可以是可靠的，但可能需要额外的逻辑来处理重复数据。如果在数据可用时抓取工具没有运行，也有可能错过数据。当没有可用的消息总线或外部回调/触发器时，这种模式很有用。

选择合适的模式

使用事件总线插件：当您的平台可以将事件发布到消息总线、需要可靠的事件传递或需要处理高容量事件时
使用回调插件：当外部系统只能通过 Webhook 推送事件，且您可以接受可靠性权衡时
使用抓取插件：当没有可用的消息总线且数据源可以定期轮询时

以下示例演示了一个完整的事件源插件模板，可作为开发自定义插件的起点。此模板包含使用 Sidecar 格式（DOCUMENTATION 和 EXAMPLES 块）的正确文档，这是插件在自动化中心（Automation Hub）中正确渲染所必需的。

"""
template.py

An ansible-rulebook event source plugin template.
"""
import asyncio
from typing import Any, Dict


DOCUMENTATION = r"""
---
short_description: A template event source plugin
description:
  - An ansible-rulebook event source plugin template.
  - This plugin generates events at a specified interval.
  - Use this as a starting point for developing custom event source plugins.
version_added: "1.0.0"
options:
  delay:
    description:
      - Number of seconds to wait between events.
    type: int
    default: 0
    required: false
  message:
    description:
      - The message to include in generated events.
    type: str
    default: "hello world"
    required: false
notes:
  - This is a template plugin for demonstration purposes.
"""

EXAMPLES = r"""
# Simple example with minimal configuration
- name: Example rulebook using template plugin
  hosts: all
  sources:
    - name: template_source
      template:
        delay: 5
        message: "custom message"

# Example with variables
- name: Using variables
  hosts: all
  sources:
    - name: template_source
      template:
        delay: "{{ event_delay }}"
        message: "{{ event_message }}"
"""


async def main(queue: asyncio.Queue, args: Dict[str, Any]):
    """
    Main plugin entrypoint.

    Args:
        queue: An asyncio queue for sending events to ansible-rulebook
        args: Plugin arguments from the rulebook
    """
    delay = args.get("delay", 0)
    message = args.get("message", "hello world")

    while True:
        await queue.put(dict(template=dict(msg=message)))
        await asyncio.sleep(delay)


if __name__ == "__main__":
    # Test mode - allows standalone execution for development/testing
    class MockQueue:
        async def put(self, event):
            print(event)

    mock_arguments = dict(delay=1, message="test message")
    asyncio.run(main(MockQueue(), mock_arguments))

插件 Python 文件必须包含一个完全如下所示的入口函数：

async def main(queue: asyncio.Queue, args: Dict[str, Any]):

这是一个异步函数。第一个参数是一个将由 ansible-rulebook CLI 消费的 asyncio 队列。其余参数是自定义定义的。它们必须与规则手册（rulebook）的 source 部分中的参数匹配。例如，模板插件期望像 delay 和 message 这样的参数。在规则手册中，source 部分如下所示：

- name: example
  hosts: all
  sources:
    - template:
        delay: 5
        message: "hello world"

每个源必须包含一个作为插件名称的键。其嵌套键必须匹配主函数期望的参数名称。插件名称即 Python 文件名。如果插件来自某个集合，则插件名称是 FQCN（完全限定集合名称），即集合名称与 Python 文件名通过点号连接，例如 ansible.eda.alertmanager。

在主函数中，您可以实现连接到外部事件源、检索事件并将它们放入提供的 asyncio 队列的代码。放入队列的事件数据必须是一个字典。您可以插入 meta 键，它指向另一个包含主机列表的字典。这些主机将限制 Ansible Playbook 的运行位置。一个简单的示例如下：{"i": 2, "meta": {"hosts": "localhost"}}。hosts 可以是逗号分隔的字符串或主机名列表。

由于插件将事件放入由 ansible-rulebook 消费的有界队列（maxsize=1）中，我们建议始终使用 await queue.put(data) 方法来放入事件，因为它会在队列满时等待，直到有可用空间。为了给事件循环提供空闲 CPU 周期来处理事件，我们建议在 put 方法后立即使用 asyncio.sleep(0)。

注意

ansible-rulebook 旨在作为一个长时间运行的进程，并随时间推移对事件做出反应。如果 任何源 的 main 函数退出，ansible-rulebook 进程将被终止。通常，您可能需要实现一个不断运行并无休止等待事件的循环。

注意

规则手册可以通过 shutdown 动作包含其自己的逻辑来结束进程。如果您的插件需要在进程终止前执行某些清理工作，必须捕获 asyncio.CancelledError 异常。

注意

请注意在插件中处理错误，并确保抛出一个带有明确消息的异常，以便 ansible-rulebook 能够正确记录它。

以下是一些测试插件的方法：

独立测试

推荐的方法是在插件文件中包含一个 if __name__ == "__main__": 块，允许它独立运行进行测试。这在上面的插件模板中已展示。

if __name__ == "__main__":
    # Test mode - allows standalone execution for development/testing
    class MockQueue:
        async def put(self, event):
            print(event)

    mock_arguments = dict(delay=1, message="test message")
    asyncio.run(main(MockQueue(), mock_arguments))

然后可以直接测试插件。

$ python3 extensions/eda/plugins/event_source/my_plugin.py

使用规则手册进行测试

创建一个使用该插件进行各种配置的测试规则手册。

- name: Test my custom plugin
  hosts: localhost
  sources:
    - name: test_source
      my_namespace.my_collection.my_plugin:
        param1: value1
        param2: value2
  rules:
    - name: Debug events
      condition: event.my_plugin is defined
      action:
        debug:
          msg: "Received event: {{ event }}"

然后使用该插件运行规则手册。

$ ansible-rulebook -i inventory.yml --rulebook test_rulebook.yml -S /path/to/plugin/directory

单元测试

为了进行更全面的测试，建议使用 pytest。这是一个示例测试结构：

import asyncio
import contextlib
import pytest
from extensions.eda.plugins.event_source import my_plugin


@pytest.mark.asyncio
async def test_plugin_generates_events():
    queue = asyncio.Queue()
    args = {"delay": 0, "message": "test"}

    # Run plugin for limited time
    task = asyncio.create_task(my_plugin.main(queue, args))
    await asyncio.sleep(0.1)
    task.cancel()

    # Verify events were generated
    assert not queue.empty()
    event = await queue.get()
    assert "my_plugin" in event


@pytest.mark.asyncio
async def test_plugin_handles_invalid_args():
    queue = asyncio.Queue()
    args = {"invalid_param": "value"}

    # Plugin should handle missing required args gracefully
    # Start plugin as background task to avoid blocking
    task = asyncio.create_task(my_plugin.main(queue, args))

    try:
        # Use a short timeout to force prompt failure
        with pytest.raises(Exception):
            await asyncio.wait_for(task, timeout=1.0)
    finally:
        # Ensure task is cancelled if still running
        if not task.done():
            task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await task

对于本地测试，插件源文件可以保存到 ansible-rulebook CLI 中 -S 参数指定的文件夹下。分发和安装插件的推荐方法是通过集合。在这种情况下，插件源文件应放在 extensions/eda/plugins/event_source 文件夹下，并通过 FQCN 引用。以下规则手册示例说明了如何引用作为内置提供的 range 插件：

- name: example2
  hosts: localhost
  sources:
    - name: range
      eda.builtin.range:
        limit: 5

无论插件是本地的还是来自集合，自定义插件所需的任何依赖包都应安装在 ansible-rulebook CLI 环境中。

事件源插件必须使用带有 DOCUMENTATION 和 EXAMPLES 块的 Sidecar 文档格式。这种格式使插件文档能够在自动化中心（Automation Hub）和 Galaxy 中正确渲染。

必需的文档块

插件必须在文件顶部包含以下文档块：

DOCUMENTATION 块 - 一个描述插件、其选项和元数据的 YAML 格式字符串
EXAMPLES 块 - 展示如何在规则手册中使用插件的实用示例

DOCUMENTATION 块格式

DOCUMENTATION 块必须是一个包含 YAML 字符串的模块级变量。

DOCUMENTATION = r"""
---
short_description: A template event source plugin
description:
  - An ansible-rulebook event source plugin template.
  - This plugin generates events at a specified interval.
  - Use this as a starting point for developing custom event source plugins.
version_added: "1.0.0"
options:
  delay:
    description:
      - Number of seconds to wait between events.
    type: int
    default: 0
    required: false
  message:
    description:
      - The message to include in generated events.
    type: str
    default: "hello world"
    required: false
notes:
  - This is a template plugin for demonstration purposes.
"""

关键字段说明：

short_description：单行摘要（必需）
description：作为字符串列表的详细解释（必需）
options：所有插件参数的字典（如果插件接受参数则必需）
- description (str)：参数的作用（必需）
- type (str)：数据类型 (str, int, bool, list, dict, float, path)（必需）
- required (bool)：参数是否强制（可选）
- default：如果未提供时的默认值（可选）
- choices (list)：有效值列表（可选）
- elements (str)：当类型为 list 时的列表元素类型（可选）

EXAMPLES 块格式

EXAMPLES 块展示如何在规则手册中使用插件。

EXAMPLES = r"""
# Simple example with minimal configuration
- name: Example rulebook using template plugin
  hosts: all
  sources:
    - name: template_source
      template:
        delay: 5
        message: "custom message"

# Example with variables
- name: Using variables
  hosts: all
  sources:
    - name: template_source
      template:
        delay: "{{ event_delay }}"
        message: "{{ event_message }}"
"""

验证 (Validation)

DOCUMENTATION 和 EXAMPLES 块遵循 Ansible 模块使用的相同 YAML 格式。当您的插件在集合中分发时，这些块会自动被解析并在自动化中心和 Galaxy 中渲染，从而使您的插件文档对浏览集合的用户可用。