8 from arvados.events import subscribe
11 def main(arguments=None):
12 logger = logging.getLogger('arvados.arv-ws')
14 parser = argparse.ArgumentParser()
15 parser.add_argument('-u', '--uuid', type=str, default="", help="Filter events on object_uuid")
16 parser.add_argument('-f', '--filters', type=str, default="", help="Arvados query filter to apply to log events (JSON encoded)")
18 group = parser.add_mutually_exclusive_group()
19 group.add_argument('--poll-interval', default=15, type=int, help="If websockets is not available, specify the polling interval, default is every 15 seconds")
20 group.add_argument('--no-poll', action='store_false', dest='poll_interval', help="Do not poll if websockets are not available, just fail")
22 group = parser.add_mutually_exclusive_group()
23 group.add_argument('-p', '--pipeline', type=str, default="", help="Supply pipeline uuid, print log output from pipeline and its jobs")
24 group.add_argument('-j', '--job', type=str, default="", help="Supply job uuid, print log output from jobs")
26 args = parser.parse_args(arguments)
29 global known_component_jobs
33 known_component_jobs = set()
36 def update_subscribed_components(components):
37 global known_component_jobs
41 if "job" in components[c]:
42 pipeline_jobs.add(components[c]["job"]["uuid"])
43 if known_component_jobs != pipeline_jobs:
44 ws.unsubscribe(filters)
45 filters = [['object_uuid', 'in', [args.pipeline] + list(pipeline_jobs)]]
46 ws.subscribe([['object_uuid', 'in', [args.pipeline] + list(pipeline_jobs)]])
47 known_component_jobs = pipeline_jobs
49 api = arvados.api('v1')
52 filters += [ ['object_uuid', '=', args.uuid] ]
55 filters += json.loads(args.filters)
58 filters += [ ['object_uuid', '=', args.job] ]
61 filters += [ ['object_uuid', '=', args.pipeline] ]
68 if 'event_type' in ev and (args.pipeline or args.job):
69 if ev['event_type'] in ('stderr', 'stdout'):
70 sys.stdout.write(ev["properties"]["text"])
71 elif ev["event_type"] in ("create", "update"):
72 if ev["object_kind"] == "arvados#pipelineInstance":
73 update_subscribed_components(ev["properties"]["new_attributes"]["components"])
75 if ev["object_kind"] == "arvados#pipelineInstance" and args.pipeline:
76 if ev["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Paused"):
79 if ev["object_kind"] == "arvados#job" and args.job:
80 if ev["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Cancelled"):
82 elif 'status' in ev and ev['status'] == 200:
88 ws = subscribe(arvados.api('v1'), filters, on_message, poll_fallback=args.poll_interval)
91 c = api.pipeline_instances().get(uuid=args.pipeline).execute()
92 update_subscribed_components(c["components"])
93 if c["state"] in ("Complete", "Failed", "Paused"):
96 except KeyboardInterrupt:
98 except Exception as e: