feat: support firing of multiple rules (#565)

https://issues.redhat.com/browse/AAP-9755
https://issues.redhat.com/browse/AAP-13131

Drools has always supported firing of multiple rules when an event or a
fact comes in. Durable rules only supported this behavior for facts but
not for events. This PR adds a feature to a ruleset to indicate if it
should match_multiple rules. The default is false


https://github.com/ansible/ansible-rulebook/assets/6452699/20daa5aa-75c7-4c63-93f0-01678ba4c986

CHANGELOG.md

@@ -4,6 +4,7 @@
 ## [Unreleased]
 
 ### Added
+- support for firing multiple rules
 
 ### Fixed
 

ansible_rulebook/json_generator.py

@@ -309,6 +309,9 @@ def visit_ruleset(ruleset: RuleSet, variables: Dict):
     if ruleset.default_events_ttl:
         data["default_events_ttl"] = ruleset.default_events_ttl
 
+    if ruleset.match_multiple_rules:
+        data["match_multiple_rules"] = ruleset.match_multiple_rules
+
     return {"RuleSet": data}
 
 

ansible_rulebook/rule_types.py

@@ -78,6 +78,7 @@ class RuleSet(NamedTuple):
     gather_facts: bool
     uuid: Optional[str] = None
     default_events_ttl: Optional[str] = None
+    match_multiple_rules: bool = False
 
 
 class ActionContext(NamedTuple):

ansible_rulebook/rules_parser.py

@@ -83,6 +83,9 @@ def parse_rule_sets(
                 gather_facts=rule_set.get("gather_facts", False),
                 uuid=str(uuid.uuid4()),
                 default_events_ttl=rule_set.get("default_events_ttl", None),
+                match_multiple_rules=rule_set.get(
+                    "match_multiple_rules", False
+                ),
             )
         )
     return rule_set_list

ansible_rulebook/schema/ruleset_schema.json

@@ -25,6 +25,10 @@
                     "type": "boolean",
                     "default": false
                 },
+                "match_multiple_rules": {
+                    "type": "boolean",
+                    "default": false
+                },
                 "name": {
                     "type": "string"
                 },

docs/rulebooks.rst

@@ -16,6 +16,7 @@ A ruleset has the following properties:
 * hosts similar to Ansible playbook
 * gather_facts: boolean
 * default_events_ttl: time to keep the partially matched events around (default is 2 hours)
+* match_multiple_rules: should multiple rules be triggered for an event(default is false)
 * execution_strategy: How actions should be executed, sequential|parallel (default: sequential). For sequential strategy we wait for each action to finish before we kick off the next action.
 * sources: A list of sources
 * rules: a list of rules

tests/e2e/files/rulebooks/test_match_multiple_rules.yml

@@ -0,0 +1,17 @@
+---
+- name: Test match multiple rules
+  hosts: all
+  match_multiple_rules: true
+  sources:
+    - name: range
+      range:
+        limit: 5
+  rules:
+    - name: r1
+      condition: event.i == 1
+      action:
+        debug:
+    - name: r11
+      condition: event.i == 1
+      action:
+        print_event:

tests/e2e/test_match_multiple_rules.py

@@ -0,0 +1,89 @@
+"""
+Module with tests for websockets
+"""
+import asyncio
+import logging
+from functools import partial
+
+import pytest
+import websockets.server as ws_server
+
+from . import utils
+
+LOGGER = logging.getLogger(__name__)
+DEFAULT_TIMEOUT = 15
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_match_multiple_rules():
+    """
+    Verify that ansible-rulebook can handle rulebook
+    which matches multiple rules for a single event
+    and send the event messages to a websocket server
+    """
+    # variables
+    host = "localhost"
+    endpoint = "/api/ws2"
+    proc_id = "42"
+    port = 31415
+    rulebook = utils.BASE_DATA_PATH / "rulebooks/test_match_multiple_rules.yml"
+    websocket_address = f"ws://localhost:{port}{endpoint}"
+    cmd = utils.Command(
+        rulebook=rulebook,
+        websocket=websocket_address,
+        proc_id=proc_id,
+        heartbeat=2,
+    )
+
+    # run server and ansible-rulebook
+    queue = asyncio.Queue()
+    handler = partial(utils.msg_handler, queue=queue)
+    async with ws_server.serve(handler, host, port):
+        LOGGER.info(f"Running command: {cmd}")
+        proc = await asyncio.create_subprocess_shell(
+            str(cmd),
+            cwd=utils.BASE_DATA_PATH,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+
+        await asyncio.wait_for(proc.wait(), timeout=DEFAULT_TIMEOUT)
+        assert proc.returncode == 0
+
+    # Verify data
+    assert not queue.empty()
+
+    action_counter = 0
+    session_stats_counter = 0
+    stats = None
+    while not queue.empty():
+        data = await queue.get()
+        assert data["path"] == endpoint
+        data = data["payload"]
+
+        if data["type"] == "Action":
+            action_counter += 1
+            assert data["action"] in ("print_event", "debug")
+            assert data["action_uuid"] is not None
+            assert data["ruleset_uuid"] is not None
+            assert data["rule_uuid"] is not None
+            matching_events = data["matching_events"]
+            del matching_events["m"]["meta"]
+            assert matching_events == {"m": {"i": 1}}
+            assert data["status"] == "successful"
+
+        if data["type"] == "SessionStats":
+            session_stats_counter += 1
+            stats = data["stats"]
+            assert stats["ruleSetName"] == "Test match multiple rules"
+            assert stats["numberOfRules"] == 2
+            assert stats["numberOfDisabledRules"] == 0
+            assert data["activation_id"] == proc_id
+
+    assert stats["rulesTriggered"] == 2
+    assert stats["eventsProcessed"] == 5
+    assert stats["eventsMatched"] == 1
+
+    assert session_stats_counter >= 2
+    assert action_counter == 2

tests/examples/80_match_multiple_rules.yml

@@ -0,0 +1,17 @@
+---
+- name: 80 match multiple rules
+  hosts: all
+  match_multiple_rules: true
+  sources:
+    - name: range
+      range:
+        limit: 5
+  rules:
+    - name: r1
+      condition: event.i == 1
+      action:
+        debug:
+    - name: r11
+      condition: event.i == 1
+      action:
+        print_event:

tests/examples/81_match_single_rule.yml

@@ -0,0 +1,16 @@
+---
+- name: 81 match single rule
+  hosts: all
+  sources:
+    - name: range
+      range:
+        limit: 5
+  rules:
+    - name: r1
+      condition: event.i == 1
+      action:
+        debug:
+    - name: r11
+      condition: event.i == 1
+      action:
+        print_event:

tests/test_examples.py

@@ -2292,7 +2292,7 @@ async def test_79_workflow_job_template_exception(err_msg, err):
 
 
 @pytest.mark.asyncio
-async def test_80_workflow_job_template():
+async def test_79_workflow_job_template():
     ruleset_queues, event_log = load_rulebook(
         "examples/79_workflow_template.yml"
     )
@@ -2324,3 +2324,56 @@ async def test_80_workflow_job_template():
 
             assert action["url"] == job_url
             assert action["action"] == "run_workflow_template"
+
+
+@pytest.mark.asyncio
+async def test_80_match_multiple_rules():
+    ruleset_queues, event_log = load_rulebook(
+        "examples/80_match_multiple_rules.yml"
+    )
+
+    queue = ruleset_queues[0][1]
+    rs = ruleset_queues[0][0]
+    with SourceTask(rs.sources[0], "sources", {}, queue):
+        await run_rulesets(
+            event_log,
+            ruleset_queues,
+            dict(),
+            dict(),
+        )
+
+        checks = {
+            "max_events": 3,
+            "shutdown_events": 1,
+            "actions": [
+                "80 match multiple rules::r1::debug",
+                "80 match multiple rules::r11::print_event",
+            ],
+        }
+        await validate_events(event_log, **checks)
+
+
+@pytest.mark.asyncio
+async def test_81_match_single_rule():
+    ruleset_queues, event_log = load_rulebook(
+        "examples/81_match_single_rule.yml"
+    )
+
+    queue = ruleset_queues[0][1]
+    rs = ruleset_queues[0][0]
+    with SourceTask(rs.sources[0], "sources", {}, queue):
+        await run_rulesets(
+            event_log,
+            ruleset_queues,
+            dict(),
+            dict(),
+        )
+
+        checks = {
+            "max_events": 2,
+            "shutdown_events": 1,
+            "actions": [
+                "81 match single rule::r1::debug",
+            ],
+        }
+        await validate_events(event_log, **checks)