Merge pull request #64 from remind101/gh-54

Deal w/ CF throttling errors

stacker/providers/aws.py

@@ -5,10 +5,39 @@
 
 from . import exceptions
 from .base import BaseProvider
+from ..util import retry_with_backoff
 
 logger = logging.getLogger(__name__)
 
 
+def retry_on_throttling(fn, attempts=3, args=None, kwargs=None):
+    """Wrap retry_with_backoff to handle AWS Cloudformation Throttling.
+
+    Args:
+        fn (function): The function to call.
+        attempts (int): Maximum # of attempts to retry the function.
+        args (list): List of positional arguments to pass to the function.
+        kwargs (dict): Dict of keyword arguments to pass to the function.
+
+    Returns:
+        passthrough: This returns the result of the function call itself.
+
+    Raises:
+        passthrough: This raises any exceptions the function call raises,
+            except for boto.exception.BotoServerError, provided it doesn't
+            retry more than attempts.
+    """
+    def _throttling_checker(exc):
+        if exc.status == 400 and "Throttling" in exc.message:
+            logger.debug("AWS throttling calls.")
+            return True
+        return False
+
+    return retry_with_backoff(fn, args=args, kwargs=kwargs, attempts=attempts,
+                              exc_list=[boto.exception.BotoServerError],
+                              retry_checker=_throttling_checker)
+
+
 class Provider(BaseProvider):
     """AWS CloudFormation Provider"""
 
@@ -37,7 +66,8 @@ def cloudformation(self):
     def get_stack(self, stack_name, **kwargs):
         stack = None
         try:
-            stack = self.cloudformation.describe_stacks(stack_name)[0]
+            stack = retry_on_throttling(self.cloudformation.describe_stacks,
+                                        args=[stack_name])[0]
         except boto.exception.BotoServerError as e:
             if 'does not exist' not in e.message:
                 raise
@@ -57,30 +87,33 @@ def is_stack_destroyed(self, stack, **kwargs):
 
     def destroy_stack(self, stack, **kwargs):
         logger.info("Destroying stack: %s" % (stack.stack_name,))
-        self.cloudformation.delete_stack(stack.stack_id)
+        retry_on_throttling(self.cloudformation.delete_stack,
+                            args=[stack.stack_id])
         return True
 
     def create_stack(self, fqn, template_url, parameters, tags, **kwargs):
         logger.info("Stack %s not found, creating.", fqn)
         logger.debug("Using parameters: %s", parameters)
         logger.debug("Using tags: %s", tags)
-        self.cloudformation.create_stack(
-            fqn,
-            template_url=template_url,
-            parameters=parameters, tags=tags,
-            capabilities=['CAPABILITY_IAM'],
+        retry_on_throttling(
+            self.cloudformation.create_stack,
+            args=[fqn],
+            kwargs=dict(template_url=template_url,
+                        parameters=parameters, tags=tags,
+                        capabilities=['CAPABILITY_IAM']),
         )
         return True
 
     def update_stack(self, fqn, template_url, parameters, tags, **kwargs):
         try:
             logger.info("Attempting to update stack %s.", fqn)
-            self.cloudformation.update_stack(
-                fqn,
-                template_url=template_url,
-                parameters=parameters,
-                tags=tags,
-                capabilities=['CAPABILITY_IAM'],
+            retry_on_throttling(
+                self.cloudformation.update_stack,
+                args=[fqn],
+                kwargs=dict(template_url=template_url,
+                            parameters=parameters,
+                            tags=tags,
+                            capabilities=['CAPABILITY_IAM']),
             )
         except boto.exception.BotoServerError as e:
             if 'No updates are to be performed.' in e.message:

stacker/tests/test_util.py

@@ -7,7 +7,7 @@
 from stacker.context import Context
 from stacker.util import (
     cf_safe_name, load_object_from_string,
-    camel_to_snake, handle_hooks)
+    camel_to_snake, handle_hooks, retry_with_backoff)
 
 regions = ['us-east-1', 'cn-north-1', 'ap-northeast-1', 'eu-west-1',
            'ap-southeast-1', 'ap-southeast-2', 'us-west-2', 'us-gov-west-1',
@@ -113,3 +113,121 @@ def test_hook_failure(self):
                   'required': False}]
         # Should pass
         handle_hooks('ignore_exception', hooks, 'us-east-1', self.context)
+
+
+class TestException1(Exception):
+    pass
+
+
+class TestException2(Exception):
+    pass
+
+
+class TestExceptionRetries(unittest.TestCase):
+    def setUp(self):
+        self.counter = 0
+
+    def _works_immediately(self, a, b, x=None, y=None):
+        self.counter += 1
+        return [a, b, x, y]
+
+    def _works_second_attempt(self, a, b, x=None, y=None):
+        self.counter += 1
+        if self.counter == 2:
+            return [a, b, x, y]
+        raise Exception("Broke.")
+
+    def _second_raises_exception2(self, a, b, x=None, y=None):
+        self.counter += 1
+        if self.counter == 2:
+            return [a, b, x, y]
+        raise TestException2("Broke.")
+
+    def _throws_exception2(self, a, b, x=None, y=None):
+        self.counter += 1
+        raise TestException2("Broke.")
+
+    def test_function_works_no_retry(self):
+
+        r = retry_with_backoff(self._works_immediately,
+                               attempts=2, min_delay=0, max_delay=.1,
+                               args=['a', 'b'],
+                               kwargs={'x': 'X', 'y': 'Y'})
+        self.assertEqual(r, ['a', 'b', 'X', 'Y'])
+        self.assertEqual(self.counter, 1)
+
+    def test_retry_exception(self):
+
+        r = retry_with_backoff(self._works_second_attempt,
+                               attempts=5, min_delay=0, max_delay=.1,
+                               args=['a', 'b'],
+                               kwargs={'x': 'X', 'y': 'Y'})
+        self.assertEqual(r, ['a', 'b', 'X', 'Y'])
+        self.assertEqual(self.counter, 2)
+
+    def test_multiple_exceptions(self):
+
+        r = retry_with_backoff(self._second_raises_exception2,
+                               exc_list=(TestException1, TestException2),
+                               attempts=5, min_delay=0, max_delay=.1,
+                               args=['a', 'b'],
+                               kwargs={'x': 'X', 'y': 'Y'})
+        self.assertEqual(r, ['a', 'b', 'X', 'Y'])
+        self.assertEqual(self.counter, 2)
+
+    def test_unhandled_exception(self):
+
+        with self.assertRaises(TestException2):
+            retry_with_backoff(self._throws_exception2,
+                               exc_list=(TestException1),
+                               attempts=5, min_delay=0, max_delay=.1,
+                               args=['a', 'b'],
+                               kwargs={'x': 'X', 'y': 'Y'})
+        self.assertEqual(self.counter, 1)
+
+    def test_never_recovers(self):
+
+        with self.assertRaises(TestException2):
+            retry_with_backoff(self._throws_exception2,
+                               exc_list=(TestException1, TestException2),
+                               attempts=5, min_delay=0, max_delay=.1,
+                               args=['a', 'b'],
+                               kwargs={'x': 'X', 'y': 'Y'})
+        self.assertEqual(self.counter, 5)
+
+    def test_retry_checker(self):
+        def _throws_handled_exception(a, b, x=None, y=None):
+            self.counter += 1
+            if self.counter == 2:
+                return [a, b, x, y]
+            raise TestException2("Broke.")
+
+        def _throws_unhandled_exception(a, b, x=None, y=None):
+            self.counter += 1
+            if self.counter == 2:
+                return [a, b, x, y]
+            raise TestException2("Invalid")
+
+        def _check_for_broke_message(e):
+            if "Broke." in e.message:
+                return True
+            return False
+
+        r = retry_with_backoff(_throws_handled_exception,
+                               exc_list=(TestException2),
+                               retry_checker=_check_for_broke_message,
+                               attempts=5, min_delay=0, max_delay=.1,
+                               args=['a', 'b'],
+                               kwargs={'x': 'X', 'y': 'Y'})
+        self.assertEqual(self.counter, 2)
+        self.assertEqual(r, ['a', 'b', 'X', 'Y'])
+
+        self.counter = 0
+        with self.assertRaises(TestException2):
+            retry_with_backoff(_throws_unhandled_exception,
+                               exc_list=(TestException2),
+                               retry_checker=_check_for_broke_message,
+                               attempts=5, min_delay=0, max_delay=.1,
+                               args=['a', 'b'],
+                               kwargs={'x': 'X', 'y': 'Y'})
+        self.assertEqual(self.counter, 1)

stacker/util.py

@@ -10,7 +10,8 @@
 
 
 def retry_with_backoff(function, args=None, kwargs=None, attempts=5,
-                       min_delay=1, max_delay=3, exc_list=None):
+                       min_delay=1, max_delay=3, exc_list=None,
+                       retry_checker=None):
     """ Retries function, catching expected Exceptions. """
     args = args or []
     kwargs = kwargs or {}
@@ -23,11 +24,18 @@ def retry_with_backoff(function, args=None, kwargs=None, attempts=5,
         try:
             return function(*args, **kwargs)
         except exc_list as e:
-            if attempt == attempts:
-                logger.error("Function %s failed after %s retries. Giving up.",
-                             function.func_name, attempts)
+            # If there is no retry checker function, or if there is and it
+            # returns True, then go ahead and retry
+            if not retry_checker or retry_checker(e):
+                if attempt == attempts:
+                    logger.error("Function %s failed after %s retries. Giving "
+                                 "up.", function.func_name, attempts)
+                    raise
+                logger.debug("Caught expected exception: %r", e)
+            # If there is a retry checker function, and it returned False,
+            # do not retry
+            else:
                 raise
-            logger.debug("Caught expected exception: %r", e)
         time.sleep(sleep_time)