Merge ed1fae7 into c2d7d95

src/commons/big_query/big_query.py

@@ -7,6 +7,8 @@
 from oauth2client.client import GoogleCredentials
 
 from src.commons.decorators.cached import cached
+from src.commons.decorators.google_http_error_retry import \
+    google_http_error_retry
 from src.commons.decorators.log_time import log_time, measure_time_and_log
 from src.commons.decorators.retry import retry
 from src.commons.big_query.big_query_table import BigQueryTable
@@ -172,8 +174,7 @@ def __sync_query(self, query, timeout=30000, use_legacy_sql=False):
             projectId=configuration.backup_project_id,
             body=query_data).execute(num_retries=3)
 
-
-    @retry(HttpError, tries=6, delay=2, backoff=2)
+    @google_http_error_retry(tries=6, delay=2, backoff=2)
     def get_table(self, project_id, dataset_id, table_id, log_table=True):
         try:
             table = self.service.tables().get(
@@ -192,21 +193,15 @@ def get_table(self, project_id, dataset_id, table_id, log_table=True):
                     TableReference(project_id, dataset_id, table_id)
                 )
                 return None
-            elif ex.resp.status == 400:
-                error_message = "Received 400 error while retrieving {}" \
-                    .format(TableReference(project_id, dataset_id, table_id))
-                logging.exception(error_message)
-                return None
-            else:
-                raise ex
+            raise ex
 
     def __log_table(self, table):
         table_copy = table.copy()
         if 'schema' in table_copy:
             del table_copy['schema']
         logging.info("Table: " + json.dumps(table_copy))
 
-    @retry(HttpError, tries=6, delay=2, backoff=2)
+    @google_http_error_retry(tries=6, delay=2, backoff=2)
     def get_dataset(self, project_id, dataset_id):
         try:
             dataset = self.service.datasets().get(

src/commons/big_query/big_query_table_metadata.py

@@ -2,11 +2,10 @@
 import logging
 from types import NoneType
 
-
+from src.commons.big_query.big_query import BigQuery
 from src.commons.decorators.cached import cached
 from src.commons.error_reporting import ErrorReporting
 from src.commons.table_reference import TableReference
-from src.commons.big_query.big_query import BigQuery
 
 
 class BigQueryTableMetadata(object):
@@ -45,13 +44,13 @@ def get_table_by_reference_cached(reference):
 
     def create_the_same_empty_table(self, target_reference):
         body = {
-            "tableReference":{
-                "projectId":target_reference.get_project_id(),
-                "datasetId":target_reference.get_dataset_id(),
-                "tableId":target_reference.get_table_id(),
+            "tableReference": {
+                "projectId": target_reference.get_project_id(),
+                "datasetId": target_reference.get_dataset_id(),
+                "tableId": target_reference.get_table_id(),
             },
-            "timePartitioning":self.table_metadata.get("timePartitioning"),
-            "schema":self.table_metadata.get("schema")
+            "timePartitioning": self.table_metadata.get("timePartitioning"),
+            "schema": self.table_metadata.get("schema")
         }
         BigQuery().create_table(target_reference.get_project_id(), target_reference.get_dataset_id(), body)
 

src/commons/decorators/google_http_error_retry.py

@@ -0,0 +1,83 @@
+# Copyright (c) 2013, SaltyCrane
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#
+#  * Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the
+#    distribution.
+#
+#  * Neither the name of the SaltyCrane nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import time
+from functools import wraps
+
+import logging
+
+from apiclient.errors import HttpError
+
+
+def google_http_error_retry(tries=4, delay=3, backoff=2, logger=logging):
+    """
+    Retry calling the decorated function using an exponential backoff.
+    It retries HttpErrors with status code > 400(except 404).
+    Errors with another status code are reraised
+
+    :param tries: number of times to try (not retry) before giving up
+    :type tries: int
+    :param delay: initial delay between retries in seconds
+    :type delay: int
+    :param backoff: backoff multiplier e.g. value of 2 will double the delay
+        each retry
+    :type backoff: int
+    :param logger: logger to use. If None, print
+    :type logger: logging.Logger instance
+    """
+    def deco_retry(f):
+
+        @wraps(f)
+        def f_retry(*args, **kwargs):
+            mtries, mdelay = tries, delay
+            while mtries > 1:
+                try:
+                    return f(*args, **kwargs)
+                except HttpError as e:
+                    if e.resp.status > 400 and e.resp.status != 404:
+                        msg = "%s, Retrying '%s' because of %s args: %s " \
+                              "in %d seconds..." \
+                              % (str(e), f.__name__, type(e), e.args, mdelay)
+                        if logger:
+                            logger.warning(msg)
+                        else:
+                            print(msg) # pylint: disable=C0325
+                        time.sleep(mdelay)
+                        mtries -= 1
+                        mdelay *= backoff
+                    else:
+                        raise e
+
+            return f(*args, **kwargs)
+
+        return f_retry  # true decorator
+
+    return deco_retry

src/commons/decorators/retry.py

@@ -1,3 +1,33 @@
+# Copyright (c) 2013, SaltyCrane
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#
+#  * Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the
+#    distribution.
+#
+#  * Neither the name of the SaltyCrane nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 import time
 from functools import wraps
 

tests/big_query/test_big_query.py

@@ -41,16 +41,6 @@ def test_iterating_datasets(self):
         # then
         self.assertEqual(self.count(dataset_ids), 3)
 
-    def test_get_table_should_ignore_400_error(self):
-        # given
-        self._create_http.return_value = self.__create_get_table_400_responses()
-
-        # when
-        table = BigQuery().get_table("project_id", "dataset_id", "table_id")
-
-        # then
-        self.assertIsNone(table)
-
     def test_iterating_tables(self):
         # given
         self._create_http.return_value = self.__create_tables_list_responses()

tests/test_google_http_error_retry_decorator.py

@@ -0,0 +1,67 @@
+import unittest
+
+from apiclient.errors import HttpError
+
+from src.commons.decorators.google_http_error_retry import \
+    google_http_error_retry
+from mock import patch, mock
+
+
+class TestGoogleHttpErrorRetryDecorator(unittest.TestCase):
+
+    class TestClass(object):
+        def counter_for_400(self):
+            pass
+
+        def counter_for_404(self):
+            pass
+
+        def counter_for_500(self):
+            pass
+
+        @google_http_error_retry(tries=3, delay=0, backoff=1)
+        def raise_HttpError400(self):
+            self.counter_for_400()
+            raise HttpError(mock.Mock(status=400), 'Bad request')
+
+        @google_http_error_retry(tries=3, delay=0, backoff=1)
+        def raise_HttpError404(self):
+            self.counter_for_404()
+            raise HttpError(mock.Mock(status=404), 'Not Found')
+
+        @google_http_error_retry(tries=3, delay=0, backoff=1)
+        def raise_HttpError500(self):
+            self.counter_for_500()
+            raise HttpError(mock.Mock(status=500), 'Internal error')
+
+    @patch.object(TestClass, "counter_for_400")
+    def test_that_http_400_errors_are_not_retried(self, counter_for_400):
+        # when
+        self.execute_function_and_suppress_exceptions(
+            TestGoogleHttpErrorRetryDecorator.TestClass().raise_HttpError400)
+        # then
+        self.assertEquals(1, counter_for_400.call_count)\
+
+
+    @patch.object(TestClass, "counter_for_404")
+    def test_that_http_404_errors_are_not_retried(self, counter_for_404):
+        # when
+        self.execute_function_and_suppress_exceptions(
+            TestGoogleHttpErrorRetryDecorator.TestClass().raise_HttpError404)
+        # then
+        self.assertEquals(1, counter_for_404.call_count)
+
+    @patch.object(TestClass, "counter_for_500")
+    def test_that_http_500_errors_are_retried(self, counter_for_500):
+        # when
+        self.execute_function_and_suppress_exceptions(
+            TestGoogleHttpErrorRetryDecorator.TestClass().raise_HttpError500)
+        # then
+        self.assertEquals(3, counter_for_500.call_count)
+
+    @classmethod
+    def execute_function_and_suppress_exceptions(cls, function):
+        try:
+            function()
+        except Exception:
+            pass