4904: Add feature to arv-ws to run a command when something is added or removed from...

[arvados.git] / sdk / python / arvados / commands / ws.py

diff --git a/sdk/python/arvados/commands/ws.py b/sdk/python/arvados/commands/ws.py
index 4ac870e9c241b0d8a64d85f66dad25a94517407b..b6ad029ee88ccf466c96690877236c13de40c03a 100644 (file)
--- a/sdk/python/arvados/commands/ws.py
+++ b/sdk/python/arvados/commands/ws.py
@@ -5,8 +5,9 @@ import logging
 import argparse
 import arvados
 import json
-import time
 from arvados.events import subscribe
+import signal
+import subprocess
 
 def main(arguments=None):
     logger = logging.getLogger('arvados.arv-ws')
@@ -15,13 +16,16 @@ 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_mutually_exclusive_group('Polling fallback')
+    group = parser.add_mutually_exclusive_group()
     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_mutually_exclusive_group('Jobs and Pipelines')
+    group = parser.add_mutually_exclusive_group()
     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")
+    group.add_argument('--project', type=str, default="", help="Monitor change events for a specific project")
+
+    parser.add_argument('--command', type=str, default="", help="Command to run when project event occurs")
 
     args = parser.parse_args(arguments)
 
@@ -71,20 +75,48 @@ def main(arguments=None):
             elif ev["event_type"] in ("create", "update"):
                 if ev["object_kind"] == "arvados#pipelineInstance":
                     update_subscribed_components(ev["properties"]["new_attributes"]["components"])
-        elif 'status' in ev and ev['status'] == 200:
-            pass
-        else:
+
+                if ev["object_kind"] == "arvados#pipelineInstance" and args.pipeline:
+                    if ev["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Paused"):
+                        ws.close()
+
+                if ev["object_kind"] == "arvados#job" and args.job:
+                    if ev["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Cancelled"):
+                        ws.close()
+            return
+
+        if 'status' in ev and ev['status'] == 200:
+            return
+
+        if args.project:
+            old_attr = None
+            if 'old_attributes' in ev['properties'] and ev['properties']['old_attributes']:
+                old_attr = ev['properties']['old_attributes']
+            if args.project not in (ev['properties']['new_attributes']['owner_uuid'],
+                                    old_attr['owner_uuid'] if old_attr else None):
+                return
+
+            if args.command:
+                et = ev['event_type']
+                if ev['event_type'] == 'update' and ev['properties']['new_attributes']['owner_uuid'] != ev['properties']['old_attributes']['owner_uuid']:
+                    if args.project == ev['properties']['new_attributes']['owner_uuid']:
+                        et = 'add'
+                    else:
+                        et = 'remove'
+                subprocess.call([args.command, args.project, et, ev['object_uuid']])
+                return
+
             print json.dumps(ev)
 
     try:
-        ws = subscribe(api, filters, on_message, poll_fallback=args.poll_interval)
+        ws = subscribe(arvados.api('v1', cache=False), filters, on_message, poll_fallback=args.poll_interval)
         if ws:
             if args.pipeline:
                 c = api.pipeline_instances().get(uuid=args.pipeline).execute()
                 update_subscribed_components(c["components"])
-
-            while True:
-                signal.pause()
+                if c["state"] in ("Complete", "Failed", "Paused"):
+                    ws.close()
+            ws.run_forever()
     except KeyboardInterrupt:
         pass
     except Exception as e: