[ARVADOS] updated: 723f0c8477dbd1d6a5e90a29dd3003c9b00d04aa
git at public.curoverse.com
git at public.curoverse.com
Thu Oct 16 13:03:47 EDT 2014
Summary of changes:
sdk/python/arvados/api.py | 3 +++
sdk/python/arvados/commands/ws.py | 6 +++---
sdk/python/arvados/events.py | 21 ++++++++++++++-------
sdk/python/tests/test_websockets.py | 13 +++++++++----
4 files changed, 29 insertions(+), 14 deletions(-)
via 723f0c8477dbd1d6a5e90a29dd3003c9b00d04aa (commit)
from b3f9cef0402c67cf06f825b283d95ef5819ab93f (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 723f0c8477dbd1d6a5e90a29dd3003c9b00d04aa
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Thu Oct 16 12:23:41 2014 -0400
3609: Add additional api_ fields to api client object so that subscribe() can
create a new api client. Change test to use event variable instead of
sleeping. Fix "Web sockets" -> "Websockets". Make some command line options
mutually exclusive.
diff --git a/sdk/python/arvados/api.py b/sdk/python/arvados/api.py
index cb716f1..a8a74da 100644
--- a/sdk/python/arvados/api.py
+++ b/sdk/python/arvados/api.py
@@ -154,7 +154,10 @@ def api(version=None, cache=True, host=None, token=None, insecure=False, **kwarg
kwargs['http'] = credentials.authorize(kwargs['http'])
svc = apiclient.discovery.build('arvados', version, **kwargs)
+ svc.api_version = version
+ svc.api_host = host
svc.api_token = token
+ svc.api_insecure = insecure
kwargs['http'].cache = None
if cache:
conncache[connprofile] = svc
diff --git a/sdk/python/arvados/commands/ws.py b/sdk/python/arvados/commands/ws.py
index 457880e..4ac870e 100644
--- a/sdk/python/arvados/commands/ws.py
+++ b/sdk/python/arvados/commands/ws.py
@@ -15,11 +15,11 @@ def main(arguments=None):
parser.add_argument('-u', '--uuid', type=str, default="", help="Filter events on object_uuid")
parser.add_argument('-f', '--filters', type=str, default="", help="Arvados query filter to apply to log events (JSON encoded)")
- group = parser.add_argument_group('Polling fallback')
+ group = parser.add_mutually_exclusive_group('Polling fallback')
group.add_argument('--poll-interval', default=15, type=int, help="If websockets is not available, specify the polling interval, default is every 15 seconds")
group.add_argument('--no-poll', action='store_false', dest='poll_interval', help="Do not poll if websockets are not available, just fail")
- group = parser.add_argument_group('Jobs and Pipelines')
+ group = parser.add_mutually_exclusive_group('Jobs and Pipelines')
group.add_argument('-p', '--pipeline', type=str, default="", help="Supply pipeline uuid, print log output from pipeline and its jobs")
group.add_argument('-j', '--job', type=str, default="", help="Supply job uuid, print log output from jobs")
@@ -84,7 +84,7 @@ def main(arguments=None):
update_subscribed_components(c["components"])
while True:
- time.sleep(60)
+ signal.pause()
except KeyboardInterrupt:
pass
except Exception as e:
diff --git a/sdk/python/arvados/events.py b/sdk/python/arvados/events.py
index df64b4e..4570cc0 100644
--- a/sdk/python/arvados/events.py
+++ b/sdk/python/arvados/events.py
@@ -7,6 +7,7 @@ import ssl
import re
import config
import logging
+import arvados
_logger = logging.getLogger('arvados.events')
@@ -48,18 +49,23 @@ class PollClient(threading.Thread):
def __init__(self, api, filters, on_event, poll_time):
super(PollClient, self).__init__()
self.api = api
- self.filters = [filters]
+ if filters:
+ self.filters = [filters]
+ else:
+ self.filters = []
self.on_event = on_event
- items = self.api.logs().list(limit=1, order="id desc", filters=filters).execute()['items']
+ self.poll_time = poll_time
+ self.stop = threading.Event()
+
+ def run(self):
+ items = self.api.logs().list(limit=1, order="id desc", filters=self.filters[0]).execute()['items']
if len(items) > 0:
self.id = items[0]["id"]
else:
self.id = 0
- self.poll_time = poll_time
- self.stop = threading.Event()
+
self.on_event({'status': 200})
- def run(self):
while not self.stop.isSet():
max_id = 0
for f in self.filters:
@@ -95,10 +101,11 @@ def subscribe(api, filters, on_event, poll_fallback=15):
if ws:
ws.close_connection()
if poll_fallback:
- _logger.warn("Web sockets not available, falling back to log table polling")
+ _logger.warn("Websockets not available, falling back to log table polling")
+ api = arvados.api(version=api.api_version, cache=False, host=api.api_host, token=api.api_token, insecure=api.api_insecure)
p = PollClient(api, filters, on_event, poll_fallback)
p.start()
return p
else:
- _logger.error("Web sockets not available")
+ _logger.error("Websockets not available")
return None
diff --git a/sdk/python/tests/test_websockets.py b/sdk/python/tests/test_websockets.py
index 83b95b8..fe065e1 100644
--- a/sdk/python/tests/test_websockets.py
+++ b/sdk/python/tests/test_websockets.py
@@ -3,6 +3,7 @@ import unittest
import arvados
import arvados.events
import time
+import threading
class WebsocketTest(run_test_server.TestCaseWithServers):
MAIN_SERVER = {'websockets': True}
@@ -14,18 +15,20 @@ class WebsocketTest(run_test_server.TestCaseWithServers):
elif self.state == 2:
self.assertEqual(self.h[u'uuid'], ev[u'object_uuid'])
self.state = 3
+ self.done.set()
elif self.state == 3:
self.fail()
def runTest(self):
self.state = 1
+ self.done = threading.Event()
run_test_server.authorize_with("admin")
api = arvados.api('v1', cache=False)
- ws = arvados.events.subscribe(api, [['object_uuid', 'is_a', 'arvados#human']], lambda ev: self.on_event(ev))
+ ws = arvados.events.subscribe(api, [['object_uuid', 'is_a', 'arvados#human']], self.on_event)
time.sleep(1)
self.h = api.humans().create(body={}).execute()
- time.sleep(2)
+ self.done.wait(10)
self.assertEqual(3, self.state)
ws.close()
@@ -39,17 +42,19 @@ class PollClientTest(run_test_server.TestCaseWithServers):
elif self.state == 2:
self.assertEqual(self.h[u'uuid'], ev[u'object_uuid'])
self.state = 3
+ self.done.set()
elif self.state == 3:
self.fail()
def runTest(self):
self.state = 1
+ self.done = threading.Event()
run_test_server.authorize_with("admin")
api = arvados.api('v1', cache=False)
- ws = arvados.events.subscribe(api, [['object_uuid', 'is_a', 'arvados#human']], lambda ev: self.on_event(ev), poll_fallback=2)
+ ws = arvados.events.subscribe(api, [['object_uuid', 'is_a', 'arvados#human']], self.on_event, poll_fallback=2)
time.sleep(1)
self.h = api.humans().create(body={}).execute()
- time.sleep(5)
+ self.done.wait(10)
self.assertEqual(3, self.state)
ws.close()
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list