82 lines
3.4 KiB
Diff
82 lines
3.4 KiB
Diff
From 06e48c585c003742ff42fb1995ec18e85226055e Mon Sep 17 00:00:00 2001
|
|
From: Mitar <mitar.git@tnode.com>
|
|
Date: Mon, 11 Feb 2019 00:23:39 -0800
|
|
Subject: [PATCH 08/13] Retry watch if request expires.
|
|
|
|
---
|
|
watch/watch.py | 30 ++++++++++++++++++++++++++++--
|
|
1 file changed, 28 insertions(+), 2 deletions(-)
|
|
|
|
diff --git a/watch/watch.py b/watch/watch.py
|
|
index fe7a924..f67dbe4 100644
|
|
--- a/watch/watch.py
|
|
+++ b/watch/watch.py
|
|
@@ -12,6 +12,7 @@
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
+import http
|
|
import json
|
|
import pydoc
|
|
|
|
@@ -86,7 +87,7 @@ class Watch(object):
|
|
def unmarshal_event(self, data, return_type):
|
|
js = json.loads(data)
|
|
js['raw_object'] = js['object']
|
|
- if return_type:
|
|
+ if return_type and js['type'] != 'ERROR':
|
|
obj = SimpleNamespace(data=json.dumps(js['raw_object']))
|
|
js['object'] = self._api_client.deserialize(obj, return_type)
|
|
if hasattr(js['object'], 'metadata'):
|
|
@@ -102,6 +103,14 @@ class Watch(object):
|
|
def stream(self, func, *args, **kwargs):
|
|
"""Watch an API resource and stream the result back via a generator.
|
|
|
|
+ Note that watching an API resource can expire. The method tries to
|
|
+ resume automatically once from the last result, but if that last result
|
|
+ is too old as well, an `ApiException` exception will be thrown with
|
|
+ ``code`` 410. In that case you have to recover yourself, probably
|
|
+ by listing the API resource to obtain the latest state and then
|
|
+ watching from that state on by setting ``resource_version`` to
|
|
+ one returned from listing.
|
|
+
|
|
:param func: The API function pointer. Any parameter to the function
|
|
can be passed after this parameter.
|
|
|
|
@@ -134,6 +143,7 @@ class Watch(object):
|
|
self.resource_version = kwargs['resource_version']
|
|
|
|
timeouts = ('timeout_seconds' in kwargs)
|
|
+ retry_after_410 = False
|
|
while True:
|
|
resp = func(*args, **kwargs)
|
|
try:
|
|
@@ -141,7 +151,23 @@ class Watch(object):
|
|
# unmarshal when we are receiving events from watch,
|
|
# return raw string when we are streaming log
|
|
if watch_arg == "watch":
|
|
- yield self.unmarshal_event(line, return_type)
|
|
+ event = self.unmarshal_event(line, return_type)
|
|
+ if isinstance(event, dict) \
|
|
+ and event['type'] == 'ERROR':
|
|
+ obj = event['raw_object']
|
|
+ # Current request expired, let's retry,
|
|
+ # but only if we have not already retried.
|
|
+ if not retry_after_410 and \
|
|
+ obj['code'] == http.HTTPStatus.GONE:
|
|
+ retry_after_410 = True
|
|
+ break
|
|
+ else:
|
|
+ reason = "%s: %s" % (obj['reason'], obj['message'])
|
|
+ raise client.rest.ApiException(status=obj['code'],
|
|
+ reason=reason)
|
|
+ else:
|
|
+ retry_after_410 = False
|
|
+ yield event
|
|
else:
|
|
yield line
|
|
if self._stop:
|
|
--
|
|
2.25.1
|
|
|