[ARVADOS] updated: 89e091b3ee1fe85a68b0a9a0a619ca7baf606d2d
Git user
git at public.curoverse.com
Thu Apr 21 15:39:04 EDT 2016
Summary of changes:
sdk/python/arvados/events.py | 45 ++++++++++++++++++++++---------------
sdk/python/arvados/retry.py | 7 +++++-
sdk/python/tests/test_websockets.py | 20 +++++++++++++++++
3 files changed, 53 insertions(+), 19 deletions(-)
via 89e091b3ee1fe85a68b0a9a0a619ca7baf606d2d (commit)
from b5dc4ae38eb6da852e167952675aff59987a995c (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit 89e091b3ee1fe85a68b0a9a0a619ca7baf606d2d
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Thu Apr 21 15:39:00 2016 -0400
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 698601c..54f3019 100644
--- 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 d8f5317..dccd9c8 100644
--- 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 907dd93..475628a 100644
--- 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,
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list