Pārlūkot izejas kodu

HUE-4357 [oozie] Return the launcher task logs when submitting Oozie batch job

HUE-4323 [editor] Batch fetch_result should return the Hive section of the Oozie launcher as result

Also unifies the YARN log fetching and parsing for Pig and Oozie snippets.

Pig and Oozie returns the results portion of the logs upon completion.
Jenny Kim 9 gadi atpakaļ
vecāks
revīzija
b75c49741f

+ 72 - 0
apps/oozie/src/oozie/views/api.py

@@ -17,6 +17,7 @@
 
 import json
 import logging
+import re
 import sys
 
 from django.http import Http404
@@ -27,6 +28,8 @@ from desktop.lib.exceptions import StructuredException
 from desktop.lib.i18n import force_unicode
 from desktop.models import Document
 
+from jobbrowser.views import job_single_logs
+from jobbrowser.models import LinkJobLogs
 from oozie.forms import WorkflowForm, NodeForm, design_form_by_type
 from oozie.models import Workflow, Node, Start, End, Kill,\
                          Link, Decision, Fork, DecisionEnd, Join,\
@@ -419,3 +422,72 @@ def workflows(request):
 
 def autocomplete_properties(request):
   return JsonResponse({ 'properties': _STD_PROPERTIES })
+
+
+@error_handler
+def get_log(request, oozie_workflow, make_links=True, log_start_pattern=None, log_end_pattern=None):
+  logs = {}
+  is_really_done = False
+
+  for action in oozie_workflow.get_working_actions():
+    try:
+      if action.externalId:
+        data = job_single_logs(request, **{'job': action.externalId})
+
+        if data and 'logs' in data:
+          action_logs = data['logs'][1]
+
+          if log_start_pattern:
+            re_log_start = re.compile(log_start_pattern, re.M | re.DOTALL)
+            if re_log_start.search(action_logs):
+              action_logs = re.search(re_log_start, action_logs).group(1).strip()
+            else:
+              LOG.debug('Failed to find given start log pattern in logs: %s' % log_start_pattern)
+
+          if make_links:
+            action_logs = LinkJobLogs._make_links(action_logs)
+
+          logs[action.name] = action_logs
+
+          if log_end_pattern:
+            re_log_end = re.compile(log_end_pattern)
+            is_really_done = re_log_end.search(action_logs) is not None
+            if is_really_done and not action_logs:
+              LOG.warn('Unable to scrape full logs, try increasing the jobbrowser log_offset configuration value.')
+    except Exception, e:
+      LOG.error('An error occurred while watching the job running: %(error)s' % {'error': e})
+      is_really_done = True
+
+  workflow_actions = _get_workflow_actions(oozie_workflow, logs, is_really_done)
+
+  return logs, workflow_actions, is_really_done
+
+
+def _get_workflow_actions(oozie_workflow, logs, is_really_done=False):
+  workflow_actions = []
+
+  # Return metadata for workflow actions (required for Pig)
+  for action in oozie_workflow.get_working_actions():
+    progress = _get_progress(oozie_workflow, logs.get(action.name, ''))
+    appendable = {
+      'name': action.name,
+      'status': action.status,
+      'logs': logs.get(action.name, ''),
+      'isReallyDone': is_really_done,
+      'progress': progress,
+      'progressPercent': '%d%%' % progress,
+      'absoluteUrl': oozie_workflow.get_absolute_url(),
+    }
+    workflow_actions.append(appendable)
+
+  return workflow_actions
+
+
+def _get_progress(job, log):
+  if job.status in ('SUCCEEDED', 'KILLED', 'FAILED'):
+    return 100
+  else:
+    try:
+      return int(re.findall("MapReduceLauncher  - (1?\d?\d)% complete", log)[-1])
+    except:
+      return 0

+ 14 - 60
apps/pig/src/pig/api.py

@@ -17,7 +17,6 @@
 
 import json
 import logging
-import re
 import time
 
 from django.core.urlresolvers import reverse
@@ -25,12 +24,12 @@ from django.utils.translation import ugettext as _
 
 from desktop.lib.i18n import smart_str
 from desktop.lib.view_util import format_duration_in_millis
-from jobbrowser.views import job_single_logs
-from jobbrowser.models import LinkJobLogs
 from liboozie.oozie_api import get_oozie
 from oozie.models import Workflow, Pig
+from oozie.views.api import get_log as get_workflow_logs
 from oozie.views.editor import _submit_workflow
 
+
 LOG = logging.getLogger(__name__)
 
 
@@ -42,17 +41,20 @@ class OozieApi(object):
   """
   Oozie submission.
   """
+
   WORKFLOW_NAME = 'pig-app-hue-script'
-  RE_LOG_END = re.compile('(<<< Invocation of Pig command completed <<<|<<< Invocation of Main class completed <<<)')
-  RE_LOG_START_RUNNING = re.compile('(Pig script \[(?:[\w.-]+)\] content:.+)', re.M | re.DOTALL)
+  LOG_START_PATTERN = '(Pig script \[(?:[\w.-]+)\] content:.+)'
+  LOG_END_PATTERN = '(<<< Invocation of Pig command completed <<<|<<< Invocation of Main class completed <<<)'
   MAX_DASHBOARD_JOBS = 100
 
+
   def __init__(self, fs, jt, user):
     self.oozie_api = get_oozie(user)
     self.fs = fs
     self.jt = jt
     self.user = user
 
+
   def submit(self, pig_script, params):
     workflow = None
 
@@ -66,6 +68,7 @@ class OozieApi(object):
 
     return oozie_wf
 
+
   def _create_workflow(self, pig_script, params):
     workflow = Workflow.objects.new_workflow(self.user)
     workflow.schema_version = 'uri:oozie:workflow:0.5'
@@ -139,6 +142,7 @@ class OozieApi(object):
 
     return workflow
 
+
   def _build_parameters(self, params):
     pig_params = []
 
@@ -154,9 +158,11 @@ class OozieApi(object):
 
     return pig_params
 
+
   def stop(self, job_id):
     return self.oozie_api.job_control(job_id, 'kill')
 
+
   def get_jobs(self):
     kwargs = {'cnt': OozieApi.MAX_DASHBOARD_JOBS,}
     kwargs['filters'] = [
@@ -166,54 +172,11 @@ class OozieApi(object):
 
     return self.oozie_api.get_workflows(**kwargs).jobs
 
-  def get_log(self, request, oozie_workflow):
-    logs = {}
-    is_really_done = False
-
-    for action in oozie_workflow.get_working_actions():
-      try:
-        if action.externalId:
-          data = job_single_logs(request, **{'job': action.externalId})
-
-          if data and 'logs' in data:
-            matched_logs = self._match_logs(data)
-
-            if matched_logs:
-              logs[action.name] = LinkJobLogs._make_links(matched_logs)
-
-            is_really_done = OozieApi.RE_LOG_END.search(data['logs'][1]) is not None
-            if is_really_done and not matched_logs:
-              LOG.warn('Unable to scrape full pig logs, try increasing the jobbrowser log_offset configuration value.')
-      except Exception, e:
-        LOG.error('An error occurred while watching the job running: %(error)s' % {'error': e})
-        is_really_done = True
-
-    workflow_actions = []
-
-    # Only one Pig action
-    for action in oozie_workflow.get_working_actions():
-      progress = get_progress(oozie_workflow, logs.get(action.name, ''))
-      appendable = {
-        'name': action.name,
-        'status': action.status,
-        'logs': logs.get(action.name, ''),
-        'isReallyDone': is_really_done,
-        'progress': progress,
-        'progressPercent': '%d%%' % progress,
-        'absoluteUrl': oozie_workflow.get_absolute_url(),
-      }
-      workflow_actions.append(appendable)
 
-    return logs, workflow_actions, is_really_done
+  def get_log(self, request, oozie_workflow, make_links=True):
+    return get_workflow_logs(request, oozie_workflow, make_links=make_links, log_start_pattern=self.LOG_START_PATTERN,
+                             log_end_pattern=self.LOG_END_PATTERN)
 
-  def _match_logs(self, data):
-    """Difficult to match multi lines of text"""
-    logs = data['logs'][1]
-
-    if OozieApi.RE_LOG_START_RUNNING.search(logs):
-      return re.search(OozieApi.RE_LOG_START_RUNNING, logs).group(1).strip()
-    else:
-      return None
 
   def massaged_jobs_for_json(self, request, oozie_jobs, hue_jobs):
     jobs = []
@@ -265,15 +228,6 @@ class OozieApi(object):
 
     return jobs
 
-def get_progress(job, log):
-  if job.status in ('SUCCEEDED', 'KILLED', 'FAILED'):
-    return 100
-  else:
-    try:
-      return int(re.findall("MapReduceLauncher  - (1?\d?\d)% complete", log)[-1])
-    except:
-      return 0
-
 
 def format_time(st_time):
   if st_time is None:

+ 53 - 4
desktop/libs/notebook/src/notebook/connectors/oozie_batch.py

@@ -16,7 +16,10 @@
 # limitations under the License.
 
 import logging
+import re
+import time
 
+from django.http import QueryDict
 from django.utils.translation import ugettext as _
 
 from desktop.lib.exceptions_renderable import PopupException
@@ -30,6 +33,7 @@ LOG = logging.getLogger(__name__)
 
 try:
   from oozie.models2 import Workflow, WorkflowBuilder
+  from oozie.views.api import get_log as get_workflow_logs
   from oozie.views.dashboard import check_job_access_permission, check_job_edition_permission
   from oozie.views.editor2 import _submit_workflow
 except Exception, e:
@@ -38,12 +42,17 @@ except Exception, e:
 
 class OozieApi(Api):
 
+  LOG_START_PATTERN = '(>>> Invoking Main class now >>>.+)'
+  LOG_END_PATTERN = '<<< Invocation of Main class completed <<<'
+  RESULTS_PATTERN = "(?P<results>>>> Invoking Beeline command line now >>>.+<<< Invocation of Beeline command completed <<<)"
+
   def __init__(self, *args, **kwargs):
     Api.__init__(self, *args, **kwargs)
 
     self.fs = self.request.fs
     self.jt = self.request.jt
 
+
   def execute(self, notebook, snippet):
     # Get document from notebook
     if not notebook.get('uuid', ''):
@@ -67,6 +76,7 @@ class OozieApi(Api):
       'has_result_set': True,
     }
 
+
   def check_status(self, notebook, snippet):
     response = {}
     job_id = snippet['result']['handle']['id']
@@ -79,16 +89,19 @@ class OozieApi(Api):
 
     return response
 
+
   def fetch_result(self, notebook, snippet, rows, start_over):
-    output = self.get_log(notebook, snippet)
+    log_output = self.get_log(notebook, snippet)
+    results = self._get_results(log_output)
 
     return {
-        'data':  [[line] for line in output.split('\n')], # hdfs_link()
+        'data':  [[line] for line in results.split('\n')],  # hdfs_link()
         'meta': [{'name': 'Header', 'type': 'STRING_TYPE', 'comment': ''}],
         'type': 'table',
         'has_more': False,
     }
 
+
   def cancel(self, notebook, snippet):
     job_id = snippet['result']['handle']['id']
 
@@ -99,13 +112,13 @@ class OozieApi(Api):
 
     return {'status': 0}
 
+
   def get_log(self, notebook, snippet, startFrom=0, size=None):
     job_id = snippet['result']['handle']['id']
 
     oozie_job = check_job_access_permission(self.request, job_id)
-    status_resp = oozie_job.log
+    return self._get_log_output(oozie_job)
 
-    return status_resp
 
   def progress(self, snippet, logs):
     job_id = snippet['result']['handle']['id']
@@ -113,8 +126,44 @@ class OozieApi(Api):
     oozie_job = check_job_access_permission(self.request, job_id)
     return oozie_job.get_progress(),
 
+
   def close_statement(self, snippet):
     pass
 
+
   def close_session(self, session):
     pass
+
+
+  def _get_log_output(self, oozie_workflow):
+    log_output = ''
+
+    q = QueryDict(self.request.GET, mutable=True)
+    q['format'] = 'python'  # Hack for triggering the good section in single_task_attempt_logs
+    self.request.GET = q
+
+    logs, workflow_actions, is_really_done = get_workflow_logs(self.request, oozie_workflow, make_links=False,
+                                                                 log_start_pattern=self.LOG_START_PATTERN,
+                                                                 log_end_pattern=self.LOG_END_PATTERN)
+
+    if len(logs) > 0:
+      log_output = logs.values()[0]
+      if log_output.startswith('Unable to locate'):
+        LOG.debug('Failed to get job attempt logs, possibly due to YARN archiving job to JHS. Will sleep and try again.')
+        time.sleep(5.0)
+        logs, workflow_actions, is_really_done = get_workflow_logs(self.request, oozie_workflow, make_links=False,
+                                                                   log_start_pattern=self.LOG_START_PATTERN,
+                                                                   log_end_pattern=self.LOG_END_PATTERN)
+        if len(logs) > 0:
+          log_output = logs.values()[0]
+
+    return log_output
+
+
+
+  def _get_results(self, log_output):
+    results = ''
+    re_results = re.compile(self.RESULTS_PATTERN, re.M | re.DOTALL)
+    if re_results.search(log_output):
+      results = re.search(re_results, log_output).group('results').strip()
+    return results

+ 51 - 17
desktop/libs/notebook/src/notebook/connectors/pig_batch.py

@@ -17,6 +17,8 @@
 
 import logging
 import json
+import re
+import time
 
 from django.core.urlresolvers import reverse
 from django.http import QueryDict
@@ -31,6 +33,7 @@ LOG = logging.getLogger(__name__)
 try:
   from pig import api
   from pig.models import PigScript2, get_workflow_output, hdfs_link
+  from oozie.views.api import get_log as get_workflow_logs
   from oozie.views.dashboard import check_job_access_permission, check_job_edition_permission
 except Exception, e:
   LOG.exception('Pig application is not enabled: %s' % e)
@@ -38,12 +41,15 @@ except Exception, e:
 
 class PigApi(Api):
 
+  RESULTS_PATTERN = "(?P<results>>>> Invoking Pig command line now >>>.+<<< Invocation of Pig command completed <<<)"
+
   def __init__(self, *args, **kwargs):
     Api.__init__(self, *args, **kwargs)
 
     self.fs = self.request.fs
     self.jt = self.request.jt
 
+
   def execute(self, notebook, snippet):
 
     attrs = {
@@ -65,16 +71,18 @@ class PigApi(Api):
       'has_result_set': True,
     }
 
+
   def check_status(self, notebook, snippet):
     job_id = snippet['result']['handle']['id']
 
     oozie_workflow = check_job_access_permission(self.request, job_id)
-    logs, workflow_actions, is_really_done = self._get_output(oozie_workflow)
+    logs, workflow_actions, is_really_done = self._get_log_output(oozie_workflow)
+    results = self._get_results(logs)
 
     if is_really_done and not oozie_workflow.is_running():
       if oozie_workflow.status in ('KILLED', 'FAILED'):
         raise QueryError(_('The script failed to run and was stopped'))
-      if logs:
+      if results:
         status = 'available'
       else:
         status = 'running' # Tricky case when the logs are being moved by YARN at job completion
@@ -87,30 +95,21 @@ class PigApi(Api):
         'status': status
     }
 
-  def _get_output(self, oozie_workflow):
-    q = QueryDict(self.request.GET, mutable=True)
-    q['format'] = 'python' # Hack for triggering the good section in single_task_attempt_logs
-    self.request.GET = q
-
-    logs, workflow_actions, is_really_done = api.get(self.fs, self.jt, self.user).get_log(self.request, oozie_workflow)
-
-    return logs, workflow_actions, is_really_done
 
   def fetch_result(self, notebook, snippet, rows, start_over):
     job_id = snippet['result']['handle']['id']
 
-    oozie_workflow = check_job_access_permission(self.request, job_id)
-    logs, workflow_actions, is_really_done = self._get_output(oozie_workflow)
-
-    output = logs.get('pig', _('No result'))
+    log_output = self.get_log(notebook, snippet)
+    results = self._get_results(log_output)
 
     return {
-        'data':  [[line] for line in output.split('\n')], # hdfs_link()
+        'data':  [[line] for line in results.split('\n')], # hdfs_link()
         'meta': [{'name': 'Header', 'type': 'STRING_TYPE', 'comment': ''}],
         'type': 'table',
         'has_more': False,
     }
 
+
   def cancel(self, notebook, snippet):
     job_id = snippet['result']['handle']['id']
 
@@ -121,13 +120,14 @@ class PigApi(Api):
 
     return {'status': 0}
 
+
   def get_log(self, notebook, snippet, startFrom=0, size=None):
     job_id = snippet['result']['handle']['id']
 
     oozie_workflow = check_job_access_permission(self.request, job_id)
-    logs, workflow_actions, is_really_done = self._get_output(oozie_workflow)
+    logs, workflow_actions, is_really_done = self._get_log_output(oozie_workflow)
+    return logs
 
-    return logs.get('pig', _('No result'))
 
   def progress(self, snippet, logs):
     job_id = snippet['result']['handle']['id']
@@ -135,8 +135,42 @@ class PigApi(Api):
     oozie_workflow = check_job_access_permission(self.request, job_id)
     return oozie_workflow.get_progress(),
 
+
   def close_statement(self, snippet):
     pass
 
+
   def close_session(self, session):
     pass
+
+
+  def _get_log_output(self, oozie_workflow):
+    log_output = ''
+
+    q = QueryDict(self.request.GET, mutable=True)
+    q['format'] = 'python'  # Hack for triggering the good section in single_task_attempt_logs
+    self.request.GET = q
+
+    logs, workflow_actions, is_really_done = api.get(self.fs, self.jt, self.user).get_log(self.request, oozie_workflow,
+                                                                                          make_links=False)
+
+    if len(logs) > 0:
+      log_output = logs.values()[0]
+      if log_output.startswith('Unable to locate'):
+        LOG.debug('Failed to get job attempt logs, possibly due to YARN archiving job to JHS. Will sleep and try again.')
+        time.sleep(5.0)
+        logs, workflow_actions, is_really_done = api.get(self.fs, self.jt, self.user).get_log(self.request, oozie_workflow,
+                                                                                              make_links=False)
+        if len(logs) > 0:
+          log_output = logs.values()[0]
+
+    return log_output, workflow_actions, is_really_done
+
+
+  def _get_results(self, log_output):
+    results = ''
+    re_results = re.compile(self.RESULTS_PATTERN, re.M | re.DOTALL)
+    if re_results.search(log_output):
+      results = re.search(re_results, log_output).group('results').strip()
+    return results
+