Cloudflare: Add Support for Rate Limit

octodns/provider/cloudflare.py

@@ -9,6 +9,7 @@ from collections import defaultdict
 from copy import deepcopy
 from logging import getLogger
 from requests import Session
+from time import sleep
 
 from ..record import Record, Update
 from .base import BaseProvider
@@ -18,7 +19,7 @@ class CloudflareError(Exception):
     def __init__(self, data):
         try:
             message = data['errors'][0]['message']
-        except (IndexError, KeyError):
+        except (IndexError, KeyError, TypeError):
             message = 'Cloudflare error'
         super(CloudflareError, self).__init__(message)
 
@@ -28,6 +29,11 @@ class CloudflareAuthenticationError(CloudflareError):
         CloudflareError.__init__(self, data)
 
 
+class CloudflareRateLimitError(CloudflareError):
+    def __init__(self, data):
+        CloudflareError.__init__(self, data)
+
+
 _PROXIABLE_RECORD_TYPES = {'A', 'AAAA', 'ALIAS', 'CNAME'}
 
 
@@ -47,6 +53,11 @@ class CloudflareProvider(BaseProvider):
         #
         # See: https://support.cloudflare.com/hc/en-us/articles/115000830351
         cdn: false
+        # Optional. Default: 4. Number of times to retry if a 429 response
+        # is received.
+        retry_count: 4
+        # Optional. Default: 300. Number of seconds to wait before retrying.
+        retry_period: 300
 
     Note: The "proxied" flag of "A", "AAAA" and "CNAME" records can be managed
           via the YAML provider like so:
@@ -66,7 +77,8 @@ class CloudflareProvider(BaseProvider):
     MIN_TTL = 120
     TIMEOUT = 15
 
-    def __init__(self, id, email=None, token=None, cdn=False, *args, **kwargs):
+    def __init__(self, id, email=None, token=None, cdn=False, retry_count=4,
+                 retry_period=300, *args, **kwargs):
         self.log = getLogger('CloudflareProvider[{}]'.format(id))
         self.log.debug('__init__: id=%s, email=%s, token=***, cdn=%s', id,
                        email, cdn)
@@ -85,11 +97,27 @@ class CloudflareProvider(BaseProvider):
                 'Authorization': 'Bearer {}'.format(token),
             })
         self.cdn = cdn
+        self.retry_count = retry_count
+        self.retry_period = retry_period
         self._sess = sess
 
         self._zones = None
         self._zone_records = {}
 
+    def _try(self, *args, **kwargs):
+        tries = self.retry_count
+        while True:  # We'll raise to break after our tries expire
+            try:
+                return self._request(*args, **kwargs)
+            except CloudflareRateLimitError:
+                if tries <= 1:
+                    raise
+                tries -= 1
+                self.log.warn('rate limit encountered, pausing '
+                              'for %ds and trying again, %d remaining',
+                              self.retry_period, tries)
+                sleep(self.retry_period)
+
     def _request(self, method, path, params=None, data=None):
         self.log.debug('_request: method=%s, path=%s', method, path)
 
@@ -101,6 +129,8 @@ class CloudflareProvider(BaseProvider):
             raise CloudflareError(resp.json())
         if resp.status_code == 403:
             raise CloudflareAuthenticationError(resp.json())
+        if resp.status_code == 429:
+            raise CloudflareRateLimitError(resp.json())
 
         resp.raise_for_status()
         return resp.json()
@@ -111,7 +141,7 @@ class CloudflareProvider(BaseProvider):
             page = 1
             zones = []
             while page:
-                resp = self._request('GET', '/zones', params={'page': page})
+                resp = self._try('GET', '/zones', params={'page': page})
                 zones += resp['result']
                 info = resp['result_info']
                 if info['count'] > 0 and info['count'] == info['per_page']:
@@ -220,7 +250,7 @@ class CloudflareProvider(BaseProvider):
             path = '/zones/{}/dns_records'.format(zone_id)
             page = 1
             while page:
-                resp = self._request('GET', path, params={'page': page})
+                resp = self._try('GET', path, params={'page': page})
                 records += resp['result']
                 info = resp['result_info']
                 if info['count'] > 0 and info['count'] == info['per_page']:
@@ -433,7 +463,7 @@ class CloudflareProvider(BaseProvider):
         zone_id = self.zones[new.zone.name]
         path = '/zones/{}/dns_records'.format(zone_id)
         for content in self._gen_data(new):
-            self._request('POST', path, data=content)
+            self._try('POST', path, data=content)
 
     def _apply_Update(self, change):
         zone = change.new.zone
@@ -522,7 +552,7 @@ class CloudflareProvider(BaseProvider):
         path = '/zones/{}/dns_records'.format(zone_id)
         for _, data in sorted(creates.items()):
             self.log.debug('_apply_Update: creating %s', data)
-            self._request('POST', path, data=data)
+            self._try('POST', path, data=data)
 
         # Updates
         for _, info in sorted(updates.items()):
@@ -532,7 +562,7 @@ class CloudflareProvider(BaseProvider):
             path = '/zones/{}/dns_records/{}'.format(zone_id, record_id)
             self.log.debug('_apply_Update: updating %s, %s -> %s',
                            record_id, data, old_data)
-            self._request('PUT', path, data=data)
+            self._try('PUT', path, data=data)
 
         # Deletes
         for _, info in sorted(deletes.items()):
@@ -541,7 +571,7 @@ class CloudflareProvider(BaseProvider):
             path = '/zones/{}/dns_records/{}'.format(zone_id, record_id)
             self.log.debug('_apply_Update: removing %s, %s', record_id,
                            old_data)
-            self._request('DELETE', path)
+            self._try('DELETE', path)
 
     def _apply_Delete(self, change):
         existing = change.existing
@@ -554,7 +584,7 @@ class CloudflareProvider(BaseProvider):
                existing_type == record['type']:
                 path = '/zones/{}/dns_records/{}'.format(record['zone_id'],
                                                          record['id'])
-                self._request('DELETE', path)
+                self._try('DELETE', path)
 
     def _apply(self, plan):
         desired = plan.desired
@@ -569,7 +599,7 @@ class CloudflareProvider(BaseProvider):
                 'name': name[:-1],
                 'jump_start': False,
             }
-            resp = self._request('POST', '/zones', data=data)
+            resp = self._try('POST', '/zones', data=data)
             zone_id = resp['result']['id']
             self.zones[name] = zone_id
             self._zone_records[name] = {}

tests/test_octodns_provider_cloudflare.py

@@ -14,7 +14,8 @@ from unittest import TestCase
 
 from octodns.record import Record, Update
 from octodns.provider.base import Plan
-from octodns.provider.cloudflare import CloudflareProvider
+from octodns.provider.cloudflare import CloudflareProvider, \
+    CloudflareRateLimitError
 from octodns.provider.yaml import YamlProvider
 from octodns.zone import Zone
 
@@ -52,7 +53,7 @@ class TestCloudflareProvider(TestCase):
     empty = {'result': [], 'result_info': {'count': 0, 'per_page': 0}}
 
     def test_populate(self):
-        provider = CloudflareProvider('test', 'email', 'token')
+        provider = CloudflareProvider('test', 'email', 'token', retry_period=0)
 
         # Bad requests
         with requests_mock() as mock:
@@ -103,6 +104,36 @@ class TestCloudflareProvider(TestCase):
                 provider.populate(zone)
             self.assertEquals(502, ctx.exception.response.status_code)
 
+        # Rate Limit error
+        with requests_mock() as mock:
+            mock.get(ANY, status_code=429,
+                     text='{"success":false,"errors":[{"code":10100,'
+                     '"message":"More than 1200 requests per 300 seconds '
+                     'reached. Please wait and consider throttling your '
+                     'request speed"}],"messages":[],"result":null}')
+
+            with self.assertRaises(Exception) as ctx:
+                zone = Zone('unit.tests.', [])
+                provider.populate(zone)
+
+            self.assertEquals('CloudflareRateLimitError',
+                              type(ctx.exception).__name__)
+            self.assertEquals('More than 1200 requests per 300 seconds '
+                              'reached. Please wait and consider throttling '
+                              'your request speed', text_type(ctx.exception))
+
+        # Rate Limit error, unknown resp
+        with requests_mock() as mock:
+            mock.get(ANY, status_code=429, text='{}')
+
+            with self.assertRaises(Exception) as ctx:
+                zone = Zone('unit.tests.', [])
+                provider.populate(zone)
+
+            self.assertEquals('CloudflareRateLimitError',
+                              type(ctx.exception).__name__)
+            self.assertEquals('Cloudflare error', text_type(ctx.exception))
+
         # Non-existent zone doesn't populate anything
         with requests_mock() as mock:
             mock.get(ANY, status_code=200, json=self.empty)
@@ -161,7 +192,7 @@ class TestCloudflareProvider(TestCase):
         self.assertEquals(13, len(again.records))
 
     def test_apply(self):
-        provider = CloudflareProvider('test', 'email', 'token')
+        provider = CloudflareProvider('test', 'email', 'token', retry_period=0)
 
         provider._request = Mock()
 
@@ -280,7 +311,11 @@ class TestCloudflareProvider(TestCase):
 
         # we don't care about the POST/create return values
         provider._request.return_value = {}
-        provider._request.side_effect = None
+
+        # Test out the create rate-limit handling, then 9 successes
+        provider._request.side_effect = [
+            CloudflareRateLimitError('{}'),
+        ] + ([None] * 3)
 
         wanted = Zone('unit.tests.', [])
         wanted.add_record(Record.new(wanted, 'nc', {
@@ -316,7 +351,7 @@ class TestCloudflareProvider(TestCase):
         ])
 
     def test_update_add_swap(self):
-        provider = CloudflareProvider('test', 'email', 'token')
+        provider = CloudflareProvider('test', 'email', 'token', retry_period=0)
 
         provider.zone_records = Mock(return_value=[
             {
@@ -357,6 +392,7 @@ class TestCloudflareProvider(TestCase):
 
         provider._request = Mock()
         provider._request.side_effect = [
+            CloudflareRateLimitError('{}'),
             self.empty,  # no zones
             {
                 'result': {
@@ -423,7 +459,7 @@ class TestCloudflareProvider(TestCase):
     def test_update_delete(self):
         # We need another run so that we can delete, we can't both add and
         # delete in one go b/c of swaps
-        provider = CloudflareProvider('test', 'email', 'token')
+        provider = CloudflareProvider('test', 'email', 'token', retry_period=0)
 
         provider.zone_records = Mock(return_value=[
             {
@@ -464,6 +500,7 @@ class TestCloudflareProvider(TestCase):
 
         provider._request = Mock()
         provider._request.side_effect = [
+            CloudflareRateLimitError('{}'),
             self.empty,  # no zones
             {
                 'result': {
@@ -1242,3 +1279,64 @@ class TestCloudflareProvider(TestCase):
         provider = CloudflareProvider('test', token='token 123')
         headers = provider._sess.headers
         self.assertEquals('Bearer token 123', headers['Authorization'])
+
+    def test_retry_behavior(self):
+        provider = CloudflareProvider('test', token='token 123',
+                                      email='email 234', retry_period=0)
+        result = {
+            "success": True,
+            "errors": [],
+            "messages": [],
+            "result": [],
+            "result_info": {
+                "count": 1,
+                "per_page": 50
+            }
+        }
+        zone = Zone('unit.tests.', [])
+        provider._request = Mock()
+
+        # No retry required, just calls and is returned
+        provider._zones = None
+        provider._request.reset_mock()
+        provider._request.side_effect = [result]
+        self.assertEquals([], provider.zone_records(zone))
+        provider._request.assert_has_calls([call('GET', '/zones',
+                                           params={'page': 1})])
+
+        # One retry required
+        provider._zones = None
+        provider._request.reset_mock()
+        provider._request.side_effect = [
+            CloudflareRateLimitError('{}'),
+            result
+        ]
+        self.assertEquals([], provider.zone_records(zone))
+        provider._request.assert_has_calls([call('GET', '/zones',
+                                           params={'page': 1})])
+
+        # Two retries required
+        provider._zones = None
+        provider._request.reset_mock()
+        provider._request.side_effect = [
+            CloudflareRateLimitError('{}'),
+            CloudflareRateLimitError('{}'),
+            result
+        ]
+        self.assertEquals([], provider.zone_records(zone))
+        provider._request.assert_has_calls([call('GET', '/zones',
+                                           params={'page': 1})])
+
+        # # Exhaust our retries
+        provider._zones = None
+        provider._request.reset_mock()
+        provider._request.side_effect = [
+            CloudflareRateLimitError({"errors": [{"message": "first"}]}),
+            CloudflareRateLimitError({"errors": [{"message": "boo"}]}),
+            CloudflareRateLimitError({"errors": [{"message": "boo"}]}),
+            CloudflareRateLimitError({"errors": [{"message": "boo"}]}),
+            CloudflareRateLimitError({"errors": [{"message": "last"}]}),
+        ]
+        with self.assertRaises(CloudflareRateLimitError) as ctx:
+            provider.zone_records(zone)
+            self.assertEquals('last', text_type(ctx.exception))