9 from arvados.events import subscribe
12 def main(arguments=None):
13 logger = logging.getLogger('arvados.arv-ws')
15 parser = argparse.ArgumentParser()
16 parser.add_argument('-u', '--uuid', type=str, default="", help="Filter events on object_uuid")
17 parser.add_argument('-f', '--filters', type=str, default="", help="Arvados query filter to apply to log events (JSON encoded)")
19 group = parser.add_mutually_exclusive_group()
20 group.add_argument('--poll-interval', default=15, type=int, help="If websockets is not available, specify the polling interval, default is every 15 seconds")
21 group.add_argument('--no-poll', action='store_false', dest='poll_interval', help="Do not poll if websockets are not available, just fail")
23 group = parser.add_mutually_exclusive_group()
24 group.add_argument('-p', '--pipeline', type=str, default="", help="Supply pipeline uuid, print log output from pipeline and its jobs")
25 group.add_argument('-j', '--job', type=str, default="", help="Supply job uuid, print log output from jobs")
27 args = parser.parse_args(arguments)
30 global known_component_jobs
34 known_component_jobs = set()
37 def update_subscribed_components(components):
38 global known_component_jobs
42 if "job" in components[c]:
43 pipeline_jobs.add(components[c]["job"]["uuid"])
44 if known_component_jobs != pipeline_jobs:
45 ws.unsubscribe(filters)
46 filters = [['object_uuid', 'in', [args.pipeline] + list(pipeline_jobs)]]
47 ws.subscribe([['object_uuid', 'in', [args.pipeline] + list(pipeline_jobs)]])
48 known_component_jobs = pipeline_jobs
50 api = arvados.api('v1', cache=False)
53 filters += [ ['object_uuid', '=', args.uuid] ]
56 filters += json.loads(args.filters)
59 filters += [ ['object_uuid', '=', args.job] ]
62 filters += [ ['object_uuid', '=', args.pipeline] ]
69 if 'event_type' in ev and (args.pipeline or args.job):
70 if ev['event_type'] in ('stderr', 'stdout'):
71 sys.stdout.write(ev["properties"]["text"])
72 elif ev["event_type"] in ("create", "update"):
73 if ev["object_kind"] == "arvados#pipelineInstance":
74 update_subscribed_components(ev["properties"]["new_attributes"]["components"])
75 elif 'status' in ev and ev['status'] == 200:
81 ws = subscribe(api, filters, on_message, poll_fallback=args.poll_interval)
84 c = api.pipeline_instances().get(uuid=args.pipeline).execute()
85 update_subscribed_components(c["components"])
89 except KeyboardInterrupt:
91 except Exception as e: