Merge pull request #83 from mindprince/child-jobs

Add support for fetching info about hadoop jobs launched by commands.

bin/qds.py

@@ -141,10 +141,22 @@ def getlogaction(cmdclass, args):
     return 0
 
 
+def getjobsaction(cmdclass, args):
+    checkargs_id(args)
+    cmd = cmdclass.find(args.pop(0))
+    if Command.is_done(cmd.status):
+        log.info("Fetching jobs for %s, Id: %s" % (cmdclass.__name__, cmd.id))
+        print(cmdclass.get_jobs_id(cmd.id))
+        return 0
+    else:
+        log.error("Cannot fetch jobs - command Id: %s is not done. Status: %s" % (cmd.id, cmd.status))
+        return 1
+
+
 def cmdmain(cmd, args):
     cmdclass = CommandClasses[cmd]
 
-    actionset = set(["submit", "run", "check", "cancel", "getresult", "getlog"])
+    actionset = set(["submit", "run", "check", "cancel", "getresult", "getlog", "getjobs"])
     if len(args) < 1:
         sys.stderr.write("missing argument containing action\n")
         usage()

qds_sdk/commands.py

@@ -136,6 +136,24 @@ def get_log(self):
         r = conn.get_raw(log_path)
         return r.text
 
+
+    @classmethod
+    def get_jobs_id(cls, id):
+        """
+        Fetches information about the hadoop jobs which were started by this
+        command id. This information is only available for commands which have
+        completed (i.e. Status = 'done', 'cancelled' or 'error'.) Also, the
+        cluster which ran this command should be running for this information
+        to be available. Otherwise only the URL and job_id is shown.
+
+        Args:
+            `id`: command id
+        """
+        conn = Qubole.agent()
+        r = conn.get_raw(cls.element_path(id) + "/jobs")
+        return r.text
+
+
     def get_results(self, fp=sys.stdout, inline=True, delim=None):
         """
         Fetches the result for the command represented by this object

tests/test_command.py

@@ -6,7 +6,6 @@
 else:
     import unittest2 as unittest
 from mock import Mock
-import tempfile
 sys.path.append(os.path.join(os.path.dirname(__file__), '../bin'))
 import qds
 import qds_sdk
@@ -141,6 +140,28 @@ def test_dbtapquerycmd(self):
                 {'status': 'kill'})
 
 
+class TestCommandGetJobs(QdsCliTestCase):
+
+    def test_running(self):
+        sys.argv = ['qds.py', 'hivecmd', 'getjobs', '123']
+        print_command()
+        Connection._api_call = Mock(return_value={'id':123, 'status': 'running'})
+        Connection._api_call_raw = Mock()
+        qds.main()
+        Connection._api_call.assert_called_with('GET', 'commands/123', params=None),
+        assert not Connection._api_call_raw.called
+
+    def test_done(self):
+        sys.argv = ['qds.py', 'hivecmd', 'getjobs', '123']
+        print_command()
+        Connection._api_call = Mock(return_value={'id':123, 'status': "done"})
+        jobs_response = Mock(text='[{"url":"https://blah","job_stats":{},"job_id":"job_blah"}]')
+        Connection._api_call_raw = Mock(return_value=jobs_response)
+        qds.main()
+        Connection._api_call.assert_called_with('GET', 'commands/123', params=None),
+        Connection._api_call_raw.assert_called_with('GET', 'commands/123/jobs', params=None),
+
+
 class TestHiveCommand(QdsCliTestCase):
 
     def test_submit_query(self):