Creating custom Event-Driven Ansible source plugins

We're surrounded! Our modern systems and applications are constantly generating events. These events could be generated by service requests, application events, health checks, etc. With the wealth of information from event traffic surrounding everything we do, Event-Driven Ansible allows for automated responses to incoming events.

But not only are we completely engulfed in event data, we're also enveloped by event sources. Think about your organization or even your household for a minute and consider how many pieces of equipment or applications are generating data that could be put to use if only you were able to easily collect it.

Event source plugins within Event-Driven Ansible act as a bridge between Ansible and event generating applications and services. Event-Driven Ansible already has a handful of event plugins to consume events from a variety of sources. But what if your source plug-in isn't represented in that list? Or what if you're a Red Hat partner who wants to connect Event-Driven Ansible to your own solution? The good news is, developing event source plugins for Event-Driven Ansible can be a relatively painless endeavor.

What is a source plugin?

Event-Driven Ansible leverages rulebooks to codify the response to an event. Rulebooks combine sources, conditions and actions. An action is executed based on one or more conditions of an event coming from a source. Event source plugins allow rulebooks to receive events from things like cloud services, applications and brokers. Without an event source, events aren't received and actions aren't taken.

Event sources are Python scripts contained within an Ansible Content Collection. Within a rulebook, event sources are called by name and parameters included in the rulebook source configuration are passed into the event source plugin. Within the event source plugin, routines should be written as asynchronous to prevent blocking, allowing events to be received and addressed as efficiently as possible across multiple event sources. For this reason, you'll notice that all of the initial source plugins like Kafka and webhook take advantage of the asynchronous IO paradigm.

Source plugin guidelines

Scoping a new event source plugin should be straightforward. For that reason, there aren't many requirements for the plugin. To get started with plugin development, here are some guidelines for source plugins:

The source plugin must contain a specific entry point.
Each source must have nested keys which match arguments expected by the main function.
Source plugins should be documented with intended purpose, expected arguments, and a rulebook example.
Event source plugins should be distributed within Collections.
Python routines should be written as non-blocking or asynchronous.
Source plugins should include a way to test the plugin outside of Event-Driven Ansible.

To demonstrate some of these guidelines, I'll use an example source plugin that I created. My source plugin is called new_records and it watches a table within ServiceNow for new records to be created (e.g. new incidents, problems and change requests). If you'd like to test this source plugin for yourself, you'll need a ServiceNow instance which you can provision as part of the ServiceNow developer program.

Before you go out and test my example plugin, please know that this plugin is coming from a sub-par python person, is meant to be an example and not at all endorsed or suggested for production use. ServiceNow instances also have rate limit rules for REST resources that you may hit by polling too often. Considering that the event push paradigm is preferred for Event-Driven Ansible source plugins, a better implementation of this source plugin might be to create a ServiceNow webservice to push event details to an event aggregator! In this scenario, our integrated application (ServiceNow) would PUSH event details to something like JetStream or Kafka (for which there is already an event source plugin!).

The source plugin must contain a specific entry point.

A source plugin requires a pretty specific entrypoint configuration. This entrypoint represents a function within the Python script that will be called by ansible-rulebook, the component of Event-Driven Ansible responsible for executing rulebooks. Let's take a look at the very beginning of my custom source plugin for ServiceNow:

import asyncio
import time
import os
from typing import Any, Dict
import aiohttp

# Entrypoint from ansible-rulebook
async def main(queue: asyncio.Queue, args: Dict[str, Any]):

After all of the import statements at the beginning of my plugin, you can see the entrypoint is an asynchronous function called main, which accepts two arguments. The first argument is an asyncio queue that will be consumed by ansible-rulebook as this source is used within a rulebook. The second argument creates a dictionary of arguments that my particular source plugin requires to make a connection to my ServiceNow instance. This dictionary will include things like username, password and URL for my ServiceNow instance. That's really all that's expected as far as the entrypoint is concerned.

Each source must have nested keys which match arguments expected by the main function.

This is a slightly more complicated way of saying that the arguments I require within my custom ServiceNow event plugin should also be keys within the rulebook used to configure the source plugin. To demonstrate this, look at the source configuration for my custom plugin within a rulebook and then look at the arguments expected by the main function that ansible-rulebook executes:

Rulebook example:

- name: Watch for new records
  hosts: localhost
  sources:
    - cloin.servicenow.new_records:
            instance: https://dev-012345.service-now.com
            username: ansible
            password: ansible
            table: incident
            interval: 1

Plugin code:

# Entrypoint from ansible-rulebook
async def main(queue: asyncio.Queue, args: Dict[str, Any]):

    instance = args.get("instance")
    username = args.get("username")
    password = args.get("password")
    table   = args.get("table")
    query   = args.get("query", "sys_created_onONToday@javascript:gs.beginningOfToday()@javascript:gs.endOfToday()")
    interval = int(args.get("interval", 5))

As a note, if you're worried about distributing rulebooks with credentials or other sensitive arguments, ansible-rulebook also accepts variables set in vars files or from environment variables using --vars or --env-vars respectively. This would mean that your rulebook source configuration could look more like:

- name: Watch for new records
  hosts: localhost
  sources:
    - cloin.servicenow.new_records:
        instance: {{ SN_HOST }}
        username: {{ SN_USERNAME }}
        password: {{ SN_PASSWORD }}
        table: incident
        interval: 1

Source plugins should be documented with purpose, expected arguments, and a rulebook example.

This is sort of a no-brainer that even I, an incredibly sub-par Python developer, can get on board with. In fact, this is actually one of my New Year's resolutions for 2023. Take a look at the top of my source plugin as an example:

"""
new_records.py

Description:
event-driven-ansible source plugin example
Poll ServiceNow API for new records in a table
Only retrieves records created after the script began executing
This script can be tested outside of ansible-rulebook by specifying
environment variables for SN_HOST, SN_USERNAME, SN_PASSWORD, SN_TABLE

Arguments:
  - instance: ServiceNow instance (e.g. https://dev-012345.service-now.com)
  - username: ServiceNow username
  - password: ServiceNow password
  - table:  Table to watch for new records
  - query:  (optional) Records to query. Defaults to records created today
  - interval: (optional) How often to poll for new records. Defaults to 5 seconds

Usage in a rulebook:
- name: Watch for new records
  hosts: localhost
  sources:
    - cloin.servicenow.new_records:
            instance: https://dev-012345.service-now.com
            username: ansible
            password: ansible
            table: incident
            interval: 1
  rules:
    - name: New record created
      condition: event.sys_id is defined
      action:
            debug:
"""

Fair enough of a guideline, right? The documentation pretty clearly lays out that this is an Event-Driven Ansible plugin, what the plugin can be expected to do, the arguments the plugin accepts and how to use this plugin within a rulebook.

Event source plugins should be distributed within Collections.

Ansible Content Collections represent the model by which Ansible content can be easily distributed. Typically, these Collections contain things like plugins, roles, playbooks and documentation, and demonstrate Ansible's extensibility. Event source plugins and rulebooks become just additional content types that can be distributed by way of Ansible Content Collections. This is demonstrated in my plugin documentation here:

- name: Watch for new records
  hosts: localhost
  sources:
    - cloin.servicenow.new_records:
            instance: https://dev-012345.service-now.com

Python routines should be written as non-blocking or asynchronous.

The asynchronous model says that, for example, requests against the ServiceNow API by the new_records source plugin shouldn't block or slow down requests to another API by another source plugin. By using asyncio along with async and await within the plugin, we simply pause that one routine and await a result instead of blocking other routines from executing. If you combine two source plugins written to utilize only synchronous routines into the same rulebook, you could find that your rulebook executes slowly or reacts to events long after they occurred. Here's an example from my source plugin:

            async with session.get(f'{instance}/api/now/table/{table}?sysparm_query={query}', auth=auth) as resp:
                if resp.status == 200:

                    records = await resp.json()
                    for record in records['result']:
…
                      await queue.put(record)

Note the keywords async and await. The async keyword lets Python know that this coroutine will be executed asynchronously within an event loop while waiting on the result from whatever has been "awaited" designated by the await keyword, in this case, the response from the ServiceNow API call.

Another line worth mentioning is the final await in the above snippet of queue.put(record). This is an essential line as this is how the record can be consumed by the rulebook engine. By putting the record returned by the ServiceNow API onto the queue, we're able to execute actions defined in the rulebook based on the record returned by the API request.

Source plugins should include a way to test the plugin outside of Event-Driven Ansible.

This one really isn't a hard and fast rule for creating source plugins. I'd say it's more helpful in the plugin development process and may more resemble a best practice or general tip than anything else. By including a function that only runs when the script is called directly by running, for example: python new_records.py, you're able to quickly test changes to the script without first setting up a rulebook and starting ansible-rulebook. For my sample plugin, I use the following:

# this is only called when testing plugin directly, without ansible-rulebook
if __name__ == "__main__":
    instance = os.environ.get('SN_HOST')
    username = os.environ.get('SN_USERNAME')
    password = os.environ.get('SN_PASSWORD')
    table   = os.environ.get('SN_TABLE')

    class MockQueue:
        async def put(self, event):
            print(event)

    asyncio.run(main(MockQueue(), {"instance": instance, "username": username, "password": password, "table": table}))

If you take a look at that code example, you can see a comment that this is really just for testing the Python script directly. If you want to test this code yourself, you can define the four environment variables (e.g. export SN_TABLE=incident...) and then execute the script. From there, open up your ServiceNow instance and create a new record in the table you're watching (in the case of SN_TABLE=incident, you'd want to create a new incident) and see that the script prints out the newly created record.