8931: Use RetryLoop to retry api calls. Add max_wait to RetryLoop. Add test

for api error retry.

diff --git a/sdk/python/arvados/events.py b/sdk/python/arvados/events.py
index 698601cb5d071a01d96cd8e5855cc1fdd1376bd8..54f3019f36b7b3e5544e3780afb730d61c44d8fd 100644 (file)
--- a/sdk/python/arvados/events.py
+++ b/sdk/python/arvados/events.py
@@ -1,9 +1,11 @@
 import arvados
 import config
 import errors
+from retry import RetryLoop
 
 import logging
 import json
+import thread
 import threading
 import time
 import os
@@ -108,7 +110,8 @@ class EventClient(object):
         try:
             self.on_event_cb(m)
         except Exception as e:
-            _logger.warn("Unexpected exception from event callback.", exc_info=e)
+            _logger.exception("Unexpected exception from event callback.")
+            thread.interrupt_main()
 
     def on_closed(self):
         if self.is_closed == False:
@@ -144,14 +147,20 @@ class PollClient(threading.Thread):
             self.id = self.last_log_id
         else:
             for f in self.filters:
-                try:
-                    items = self.api.logs().list(limit=1, order="id desc", filters=f).execute(num_retries=1000000)['items']
-                except Exception as e:
-                    # Some apparently non-retryable error happened, so log the
-                    # error and shut down gracefully.
-                    _logger.error("Got exception from log query: %s", e)
+                for tries_left in RetryLoop(num_retries=25, backoff_start=.1, max_wait=self.poll_time):
+                    try:
+                        items = self.api.logs().list(limit=1, order="id desc", filters=f).execute()['items']
+                        break
+                    except errors.ApiError as error:
+                        pass
+                    else:
+                        tries_left = 0
+                        break
+                if tries_left == 0:
+                    _logger.exception("PollClient thread could not contact API server.")
                     with self._closing_lock:
                         self._closing.set()
+                    thread.interrupt_main()
                     return
                 if items:
                     if items[0]['id'] > self.id:
@@ -163,18 +172,17 @@ class PollClient(threading.Thread):
             max_id = self.id
             moreitems = False
             for f in self.filters:
-                try:
-                    # If we get a transient error, we really really need to
-                    # just keep trying over and over with the same query or
-                    # we'll potentially drop events which would break the event
-                    # stream contract.
-                    items = self.api.logs().list(order="id asc", filters=f+[["id", ">", str(self.id)]]).execute(num_retries=1000000)
-                except Exception as e:
-                    # Some apparently non-retryable error happened, so log the
-                    # error and shut down gracefully.
-                    _logger.error("Got exception from log query: %s", e)
+                for tries_left in RetryLoop(num_retries=25, backoff_start=.1, max_wait=self.poll_time):
+                    try:
+                        items = self.api.logs().list(order="id asc", filters=f+[["id", ">", str(self.id)]]).execute()
+                        break
+                    except errors.ApiError as error:
+                        pass
+                if tries_left == 0:
+                    _logger.exception("PollClient thread could not contact API server.")
                     with self._closing_lock:
                         self._closing.set()
+                    thread.interrupt_main()
                     return
                 for i in items["items"]:
                     if i['id'] > max_id:
@@ -185,7 +193,8 @@ class PollClient(threading.Thread):
                         try:
                             self.on_event(i)
                         except Exception as e:
-                            _logger.warn("Unexpected exception from event callback.", exc_info=e)
+                            _logger.exception("Unexpected exception from event callback.")
+                            thread.interrupt_main()
                 if items["items_available"] > len(items["items"]):
                     moreitems = True
             self.id = max_id

diff --git a/sdk/python/arvados/retry.py b/sdk/python/arvados/retry.py
index d8f5317d2c4c160c833339929302edaae679d6f5..dccd9c875a69ee161923e6fa6818e2d751f92352 100644 (file)
--- a/sdk/python/arvados/retry.py
+++ b/sdk/python/arvados/retry.py
@@ -31,7 +31,8 @@ class RetryLoop(object):
             return loop.last_result()
     """
     def __init__(self, num_retries, success_check=lambda r: True,
-                 backoff_start=0, backoff_growth=2, save_results=1):
+                 backoff_start=0, backoff_growth=2, save_results=1,
+                 max_wait=60):
         """Construct a new RetryLoop.
 
         Arguments:
@@ -50,11 +51,13 @@ class RetryLoop(object):
         * save_results: Specify a number to save the last N results
           that the loop recorded.  These records are available through
           the results attribute, oldest first.  Default 1.
+        * max_wait: Maximum time to wait between retries.
         """
         self.tries_left = num_retries + 1
         self.check_result = success_check
         self.backoff_wait = backoff_start
         self.backoff_growth = backoff_growth
+        self.max_wait = max_wait
         self.next_start_time = 0
         self.results = deque(maxlen=save_results)
         self._running = None
@@ -76,6 +79,8 @@ class RetryLoop(object):
             wait_time = max(0, self.next_start_time - time.time())
             time.sleep(wait_time)
             self.backoff_wait *= self.backoff_growth
+            if self.backoff_wait > self.max_wait:
+                self.backoff_wait = self.max_wait
         self.next_start_time = time.time() + self.backoff_wait
         self.tries_left -= 1
         return self.tries_left

diff --git a/sdk/python/tests/test_websockets.py b/sdk/python/tests/test_websockets.py
index 907dd93100e645082b06a55841fffcdc0ff350f4..475628ac11be3855858ac4d4ddea8c7dc968da8a 100644 (file)
--- a/sdk/python/tests/test_websockets.py
+++ b/sdk/python/tests/test_websockets.py
@@ -1,5 +1,6 @@
 import arvados
 import arvados.events
+import arvados.errors
 from datetime import datetime, timedelta, tzinfo
 import logging
 import logging.handlers
@@ -78,6 +79,25 @@ class WebsocketTest(run_test_server.TestCaseWithServers):
         self._test_subscribe(
             poll_fallback=0.25, expect_type=arvados.events.PollClient, expected=1)
 
+    def test_subscribe_poll_retry(self):
+        api_mock = mock.MagicMock()
+        n = []
+        def on_ev(ev):
+            n.append(ev)
+
+        error_mock = mock.MagicMock()
+        error_mock.resp.status = 0
+        error_mock._get_reason.return_value = "testing"
+        api_mock.logs().list().execute.side_effect = (arvados.errors.ApiError(error_mock, ""),
+                                                      {"items": [{"id": 1}], "items_available": 1},
+                                                      arvados.errors.ApiError(error_mock, ""),
+                                                      {"items": [{"id": 1}], "items_available": 1})
+        pc = arvados.events.PollClient(api_mock, [], on_ev, 15, None)
+        pc.start()
+        while len(n) < 2:
+            time.sleep(.1)
+        pc.close()
+
     def test_subscribe_websocket_with_start_time_past(self):
         self._test_subscribe(
             poll_fallback=False, expect_type=arvados.events.EventClient,