Add BaseWaiterFactory, so we can start splitting the waiters (#2317)

SUMMARY
Having everything in waiters.py today means that every time we add a new waiter we trigger a lot of integration tests.  Adding this Factory gives us a basis to work upon for splitting the code up.  It's based upon the Factory in community.aws, but doesn't come with the assumption that we'll have the module/client available to us when we instantiate the factory.
While similar to the current code in waiters.py, this factory supports:

waiter definitions stored on a per-API basis
waiter definitions outside of amazon.aws

Because touching the existing waiters.py would trigger a mass of integration tests, this puts the factory into waiter.py instead of waiters.py.  Additionally, putting it in waiter.py also gives us a long term route to split the current waiters off and out of waiters.py without generating import loops.  (Putting it in waiters.py would mean that a per-API file would need to import waiters.py, and thus waiters.py couldn't import the new file to provide backwards compatibility with today's implementation)
ISSUE TYPE

Feature Pull Request

COMPONENT NAME
plugin/module_utils/waiter.py
ADDITIONAL INFORMATION
See #2296 for an example of using this (module_utils._autoscaling.waiters)

Reviewed-by: Alina Buzachis

plugins/module_utils/waiter.py

@@ -0,0 +1,169 @@
+# -*- coding: utf-8 -*-
+
+# Copyright: Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+#
+# Note: This code should probably live in amazon.aws rather than community.aws.
+# However, for the sake of getting something into a useful shape first, it makes
+# sense for it to start life in community.aws.
+#
+
+# import typing
+from copy import deepcopy
+
+try:
+    import botocore.waiter as botocore_waiter
+
+    import_error = None
+except ImportError as e:
+    botocore_waiter = None
+    import_error = e
+
+
+class BaseWaiterFactory:
+    """
+    A helper class used for creating additional waiters.
+    Unlike the waiters available directly from botocore these waiters will
+    automatically retry on common (temporary) AWS failures.
+
+    This class should be treated as an abstract class and subclassed before use.
+    A subclass should:
+    - override _BaseWaiterFactory._waiter_model_data to return the data defining
+      the waiter
+
+    Usage:
+    waiter_factory = BaseWaiterFactory()
+    waiter = waiter_factory.get_waiter(client, 'my_waiter_name')
+    waiter.wait(**params)
+    """
+
+    def __init__(self):
+        if not botocore_waiter:
+            return
+
+        # While it would be nice to supliment this with the upstream data,
+        # unfortunately client doesn't have a public method for getting the
+        # waiter configs.
+        data = self._inject_ratelimit_retries(self._waiter_model_data)
+        self._model = botocore_waiter.WaiterModel(
+            waiter_config=dict(version=2, waiters=data),
+        )
+
+    @property
+    def _waiter_model_data(self):
+        r"""
+        Subclasses should override this method to return a dictionary mapping
+        waiter names to the waiter definition.
+
+        This data is similar to the data found in botocore's waiters-2.json
+        files (for example: botocore/botocore/data/ec2/2016-11-15/waiters-2.json)
+        with two differences:
+        1) Waiter names do not have transformations applied during lookup
+        2) Only the 'waiters' data is required, the data is assumed to be
+           version 2
+
+        for example:
+
+        @property
+        def _waiter_model_data(self):
+            return dict(
+                tgw_attachment_deleted=dict(
+                    operation='DescribeTransitGatewayAttachments',
+                    delay=5, maxAttempts=120,
+                    acceptors=[
+                        dict(state='retry', matcher='pathAll', expected='deleting', argument='TransitGatewayAttachments[].State'),
+                        dict(state='success', matcher='pathAll', expected='deleted', argument='TransitGatewayAttachments[].State'),
+                        dict(state='success', matcher='path', expected=True, argument='length(TransitGatewayAttachments[]) == `0`'),
+                        dict(state='success', matcher='error', expected='InvalidRouteTableID.NotFound'),
+                    ]
+                ),
+            )
+
+        or
+
+        @property
+        def _waiter_model_data(self):
+            return {
+                "instance_exists": {
+                    "delay": 5,
+                    "maxAttempts": 40,
+                    "operation": "DescribeInstances",
+                    "acceptors": [
+                        {
+                            "matcher": "path",
+                            "expected": true,
+                            "argument": "length(Reservations[]) > `0`",
+                            "state": "success"
+                        },
+                        {
+                            "matcher": "error",
+                            "expected": "InvalidInstanceID.NotFound",
+                            "state": "retry"
+                        }
+                    ]
+                },
+            }
+        """
+
+        return dict()
+
+    def _inject_ratelimit_retries(self, model_data):
+        extra_retries = [
+            "RequestLimitExceeded",
+            "Unavailable",
+            "ServiceUnavailable",
+            "InternalFailure",
+            "InternalError",
+            "TooManyRequestsException",
+            "Throttling",
+        ]
+
+        acceptors = []
+        for error in extra_retries:
+            acceptors.append(dict(state="retry", matcher="error", expected=error))
+
+        _model_data = deepcopy(model_data)
+        for waiter_name in _model_data:
+            _model_data[waiter_name]["acceptors"].extend(acceptors)
+
+        return _model_data
+
+    def get_waiter(self, client, waiter_name: str):
+        if import_error:
+            # We shouldn't get here, but if someone's trying to use this without botocore installed
+            # let's re-raise the actual import error
+            raise import_error
+
+        waiters = self._model.waiter_names
+        if waiter_name not in waiters:
+            raise NotImplementedError(f"Unable to find waiter {waiter_name}.  Available_waiters: {waiters}")
+        return botocore_waiter.create_waiter_with_client(
+            waiter_name,
+            self._model,
+            client,
+        )
+
+
+def custom_waiter_config(timeout: int, default_pause: int = 2):
+    """
+    Generates the waiter_config dict that allows configuring a custom wait_timeout
+
+    Where the pause and the timeouts aren't perfectly divisible, this will default to waiting
+    slightly longer than the configured timeout so that we give at least the timeout time for a
+    change to happen.
+    """
+
+    pause = default_pause
+
+    # Do something sensible when the user's passed a short timeout, but our default pause wouldn't
+    # have allowed any retries
+    if timeout < (default_pause * 3):
+        pause = max(1, timeout // 3)
+
+    attempts = 1 + (timeout // pause)
+
+    if (attempts - 1) * pause < timeout:
+        attempts += 1
+
+    return dict(Delay=pause, MaxAttempts=attempts)

tests/unit/module_utils/waiter/test_custom_waiter_config.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+# Copyright: Ansible Project
+#
+# This file is part of Ansible
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from copy import deepcopy
+
+import pytest
+
+from ansible_collections.amazon.aws.plugins.module_utils.waiter import custom_waiter_config
+
+# Total time = (MaxAttempts - 1) * Delay
+# There's no pause before the first attempt, or after the last attempt
+TEST_DATA = [
+    # Only performs a single attempt, no retries
+    [dict(timeout=0), {"Delay": 1, "MaxAttempts": 1}],
+    # 1 second
+    [dict(timeout=1), {"Delay": 1, "MaxAttempts": 2}],
+    # 2 seconds
+    [dict(timeout=2), {"Delay": 1, "MaxAttempts": 3}],
+    # 4 seconds
+    [dict(timeout=4), {"Delay": 1, "MaxAttempts": 5}],
+    # 6 seconds
+    [dict(timeout=6), {"Delay": 2, "MaxAttempts": 4}],
+    # 10 seconds
+    [dict(timeout=10), {"Delay": 2, "MaxAttempts": 6}],
+    # 12 seconds
+    [dict(timeout=11), {"Delay": 2, "MaxAttempts": 7}],
+    # 12 seconds
+    [dict(timeout=10, default_pause=10), {"Delay": 3, "MaxAttempts": 5}],
+    # 12 seconds
+    [dict(timeout=10, default_pause=15), {"Delay": 3, "MaxAttempts": 5}],
+    # 105 seconds
+    [dict(timeout=100, default_pause=15), {"Delay": 15, "MaxAttempts": 8}],
+    # 150 seconds
+    [dict(timeout=150, default_pause=15), {"Delay": 15, "MaxAttempts": 11}],
+]
+
+
+class TestCustomWaiterConfig:
+    def setup_method(self):
+        pass
+
+    @pytest.mark.parametrize("input_params, output_params", deepcopy(TEST_DATA))
+    def test_custom_waiter(self, input_params, output_params):
+        # Test default behaviour
+        assert custom_waiter_config(**input_params) == output_params

tests/unit/plugins/modules/test_s3_object.py

@@ -6,9 +6,6 @@
 from unittest.mock import MagicMock
 from unittest.mock import patch
 
-import botocore.exceptions
-import pytest
-
 from ansible_collections.amazon.aws.plugins.modules import s3_object
 
 module_name = "ansible_collections.amazon.aws.plugins.modules.s3_object"