feat: add wait and wait_timeout to mq_broker; closes #1879 (#1888)

feat: add wait and wait_timeout to mq_broker; closes #1879

SUMMARY
add wait and wait_timeout to wait for broker desired state. This will be used instead of polling for broker desired state though a separate module, broker_info.
ISSUE TYPE

Feature Pull Request

COMPONENT NAME
mq_broker
ADDITIONAL INFORMATION
  wait:
    description:
      - Specifies whether the module waits for the desired C(state).
    type: bool
  wait_timeout:
    description:
      - How long to wait (in seconds) for the broker to reach the desired state.
    default: 900
    type: int

Reviewed-by: Alina Buzachis

changelogs/fragments/1879-mq_broker-add-wait.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - mq_broker - add support to wait for broker state via ``wait`` and ``wait_timeout`` parameter values (https://github.com/ansible-collections/community.aws/pull/1879).

plugins/modules/mq_broker.py

@@ -124,6 +124,19 @@
       - At least one must be provided during creation.
     type: list
     elements: str
+  wait:
+    description:
+      - Specifies whether the module waits for the desired C(state).
+      - The time to wait can be controlled by setting I(wait_timeout).
+    type: bool
+    default: false
+    version_added: 7.1.0
+  wait_timeout:
+    description:
+      - How long to wait (in seconds) for the broker to reach the desired state if I(wait=true).
+    default: 900
+    type: int
+    version_added: 7.1.0
 
 extends_documentation_fragment:
   - amazon.aws.boto3
@@ -215,6 +228,9 @@
     # handled by AnsibleAWSModule
     pass
 
+from time import sleep
+from time import time
+
 from ansible.module_utils.common.dict_transformations import camel_dict_to_snake_dict
 
 from ansible_collections.amazon.aws.plugins.module_utils.modules import AnsibleAWSModule
@@ -384,22 +400,77 @@ def get_broker_info(conn, module, broker_id):
         module.fail_json_aws(e, msg="Couldn't get broker details.")
 
 
+def wait_for_status(conn, module):
+    interval_secs = 5
+    timeout = module.params.get("wait_timeout", 900)
+    broker_name = module.params.get("broker_name")
+    desired_state = module.params.get("state")
+    done = False
+
+    paginator = conn.get_paginator("list_brokers")
+    page_iterator = paginator.paginate(PaginationConfig={"MaxItems": 100, "PageSize": 100, "StartingToken": ""})
+    wait_timeout = time() + timeout
+
+    while wait_timeout > time():
+        try:
+            filtered_iterator = page_iterator.search(f"BrokerSummaries[?BrokerName == `{broker_name}`][]")
+            broker_list = list(filtered_iterator)
+
+            if module.check_mode:
+                return
+
+            if len(broker_list) < 1 and desired_state == "absent":
+                done = True
+                break
+
+            if desired_state in ["present", "rebooted"] and broker_list[0]["BrokerState"] == "RUNNING":
+                done = True
+                break
+
+            if broker_list[0]["BrokerState"] == "CREATION_FAILED":
+                break
+
+            sleep(interval_secs)
+
+        except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+            module.fail_json_aws(e, msg="Couldn't paginate brokers.")
+
+    if not done:
+        module.fail_json(msg="desired state not reached")
+
+
 def reboot_broker(conn, module, broker_id):
+    wait = module.params.get("wait")
+
     try:
-        return conn.reboot_broker(BrokerId=broker_id)
+        response = conn.reboot_broker(BrokerId=broker_id)
     except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
         module.fail_json_aws(e, msg="Couldn't reboot broker.")
 
+    if wait:
+        wait_for_status(conn, module)
+
+    return response
+
 
 def delete_broker(conn, module, broker_id):
+    wait = module.params.get("wait")
+
     try:
-        return conn.delete_broker(BrokerId=broker_id)
+        response = conn.delete_broker(BrokerId=broker_id)
     except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
         module.fail_json_aws(e, msg="Couldn't delete broker.")
 
+    if wait:
+        wait_for_status(conn, module)
+
+    return response
+
 
 def create_broker(conn, module):
     kwargs = _fill_kwargs(module)
+    wait = module.params.get("wait")
+
     if "EngineVersion" in kwargs and kwargs["EngineVersion"] == "latest":
         kwargs["EngineVersion"] = get_latest_engine_version(conn, module, kwargs["EngineType"])
     if kwargs["AuthenticationStrategy"] == "LDAP":
@@ -416,11 +487,15 @@ def create_broker(conn, module):
     changed = True
     result = conn.create_broker(**kwargs)
     #
+    if wait:
+        wait_for_status(conn, module)
+
     return {"broker": camel_dict_to_snake_dict(result, ignore_list=["Tags"]), "changed": changed}
 
 
 def update_broker(conn, module, broker_id):
     kwargs = _fill_kwargs(module, apply_defaults=False, ignore_create_params=True)
+    wait = module.params.get("wait")
     # replace name with id
     broker_name = kwargs["BrokerName"]
     del kwargs["BrokerName"]
@@ -443,6 +518,9 @@ def update_broker(conn, module, broker_id):
             api_result = conn.update_broker(**kwargs)
         #
     #
+    if wait:
+        wait_for_status(conn, module)
+
     return {"broker": result, "changed": changed}
 
 
@@ -484,6 +562,8 @@ def main():
     argument_spec = dict(
         broker_name=dict(required=True, type="str"),
         state=dict(default="present", choices=["present", "absent", "restarted"]),
+        wait=dict(default=False, type="bool"),
+        wait_timeout=dict(default=900, type="int"),
         # parameters only allowed on create
         deployment_mode=dict(choices=["SINGLE_INSTANCE", "ACTIVE_STANDBY_MULTI_AZ", "CLUSTER_MULTI_AZ"]),
         use_aws_owned_key=dict(type="bool"),

tests/integration/targets/mq/tasks/broker_tests.yml

@@ -4,6 +4,7 @@
     security_groups: "{{ broker_sg_ids.split(',') }}"
     subnet_ids: "{{ broker_subnet_ids.split(',') }}"
     tags: "{{ tags }}"
+    wait: true
   register: result
 - set_fact:
     broker_id: "{{ result.broker['broker_id'] }}"
@@ -19,20 +20,10 @@
       - ( result.changed | bool )
       - result_c1.broker['broker_id'] == broker_id
       - result_c1.broker['broker_name'] == broker_name
-      - result_c1.broker['broker_state'] == 'CREATION_IN_PROGRESS'
+      - result_c1.broker['broker_state'] == 'RUNNING'
       - ( result_c1.broker['storage_type'] | upper ) == 'EFS'
       - result_c1.broker['tags'] == tags
   when: not ansible_check_mode
-- debug:
-    msg: "Wait until broker {{ broker_name }} ({{ broker_id }}) enters running state. This may take several minutes"
-- name: wait for startup
-  mq_broker_info:
-    broker_id:      "{{ broker_id }}"
-  register: result
-  until: result.broker['broker_state'] == 'RUNNING'
-  retries: 15
-  delay:   60
-  when: not ansible_check_mode
 - name: repeat creation
   mq_broker:
     broker_name: "{{ broker_name }}"