Explorar o código

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 %!s(int64=9) %!d(string=hai) anos
pai
achega
b75c497

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

@@ -17,6 +17,7 @@
 
 
 import json
 import json
 import logging
 import logging
+import re
 import sys
 import sys
 
 
 from django.http import Http404
 from django.http import Http404
@@ -27,6 +28,8 @@ from desktop.lib.exceptions import StructuredException
 from desktop.lib.i18n import force_unicode
 from desktop.lib.i18n import force_unicode
 from desktop.models import Document
 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.forms import WorkflowForm, NodeForm, design_form_by_type
 from oozie.models import Workflow, Node, Start, End, Kill,\
 from oozie.models import Workflow, Node, Start, End, Kill,\
                          Link, Decision, Fork, DecisionEnd, Join,\
                          Link, Decision, Fork, DecisionEnd, Join,\
@@ -419,3 +422,72 @@ def workflows(request):
 
 
 def autocomplete_properties(request):
 def autocomplete_properties(request):
   return JsonResponse({ 'properties': _STD_PROPERTIES })
   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 json
 import logging
 import logging
-import re
 import time
 import time
 
 
 from django.core.urlresolvers import reverse
 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.i18n import smart_str
 from desktop.lib.view_util import format_duration_in_millis
 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 liboozie.oozie_api import get_oozie
 from oozie.models import Workflow, Pig
 from oozie.models import Workflow, Pig
+from oozie.views.api import get_log as get_workflow_logs
 from oozie.views.editor import _submit_workflow
 from oozie.views.editor import _submit_workflow
 
 
+
 LOG = logging.getLogger(__name__)
 LOG = logging.getLogger(__name__)
 
 
 
 
@@ -42,17 +41,20 @@ class OozieApi(object):
   """
   """
   Oozie submission.
   Oozie submission.
   """
   """
+
   WORKFLOW_NAME = 'pig-app-hue-script'
   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
   MAX_DASHBOARD_JOBS = 100
 
 
+
   def __init__(self, fs, jt, user):
   def __init__(self, fs, jt, user):
     self.oozie_api = get_oozie(user)
     self.oozie_api = get_oozie(user)
     self.fs = fs
     self.fs = fs
     self.jt = jt
     self.jt = jt
     self.user = user
     self.user = user
 
 
+
   def submit(self, pig_script, params):
   def submit(self, pig_script, params):
     workflow = None
     workflow = None
 
 
@@ -66,6 +68,7 @@ class OozieApi(object):
 
 
     return oozie_wf
     return oozie_wf
 
 
+
   def _create_workflow(self, pig_script, params):
   def _create_workflow(self, pig_script, params):
     workflow = Workflow.objects.new_workflow(self.user)
     workflow = Workflow.objects.new_workflow(self.user)
     workflow.schema_version = 'uri:oozie:workflow:0.5'
     workflow.schema_version = 'uri:oozie:workflow:0.5'
@@ -139,6 +142,7 @@ class OozieApi(object):
 
 
     return workflow
     return workflow
 
 
+
   def _build_parameters(self, params):
   def _build_parameters(self, params):
     pig_params = []
     pig_params = []
 
 
@@ -154,9 +158,11 @@ class OozieApi(object):
 
 
     return pig_params
     return pig_params
 
 
+
   def stop(self, job_id):
   def stop(self, job_id):
     return self.oozie_api.job_control(job_id, 'kill')
     return self.oozie_api.job_control(job_id, 'kill')
 
 
+
   def get_jobs(self):
   def get_jobs(self):
     kwargs = {'cnt': OozieApi.MAX_DASHBOARD_JOBS,}
     kwargs = {'cnt': OozieApi.MAX_DASHBOARD_JOBS,}
     kwargs['filters'] = [
     kwargs['filters'] = [
@@ -166,54 +172,11 @@ class OozieApi(object):
 
 
     return self.oozie_api.get_workflows(**kwargs).jobs
     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):
   def massaged_jobs_for_json(self, request, oozie_jobs, hue_jobs):
     jobs = []
     jobs = []
@@ -265,15 +228,6 @@ class OozieApi(object):
 
 
     return jobs
     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):
 def format_time(st_time):
   if st_time is None:
   if st_time is None:

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

@@ -16,7 +16,10 @@
 # limitations under the License.
 # limitations under the License.
 
 
 import logging
 import logging
+import re
+import time
 
 
+from django.http import QueryDict
 from django.utils.translation import ugettext as _
 from django.utils.translation import ugettext as _
 
 
 from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.exceptions_renderable import PopupException
@@ -30,6 +33,7 @@ LOG = logging.getLogger(__name__)
 
 
 try:
 try:
   from oozie.models2 import Workflow, WorkflowBuilder
   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.dashboard import check_job_access_permission, check_job_edition_permission
   from oozie.views.editor2 import _submit_workflow
   from oozie.views.editor2 import _submit_workflow
 except Exception, e:
 except Exception, e:
@@ -38,12 +42,17 @@ except Exception, e:
 
 
 class OozieApi(Api):
 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):
   def __init__(self, *args, **kwargs):
     Api.__init__(self, *args, **kwargs)
     Api.__init__(self, *args, **kwargs)
 
 
     self.fs = self.request.fs
     self.fs = self.request.fs
     self.jt = self.request.jt
     self.jt = self.request.jt
 
 
+
   def execute(self, notebook, snippet):
   def execute(self, notebook, snippet):
     # Get document from notebook
     # Get document from notebook
     if not notebook.get('uuid', ''):
     if not notebook.get('uuid', ''):
@@ -67,6 +76,7 @@ class OozieApi(Api):
       'has_result_set': True,
       'has_result_set': True,
     }
     }
 
 
+
   def check_status(self, notebook, snippet):
   def check_status(self, notebook, snippet):
     response = {}
     response = {}
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
@@ -79,16 +89,19 @@ class OozieApi(Api):
 
 
     return response
     return response
 
 
+
   def fetch_result(self, notebook, snippet, rows, start_over):
   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 {
     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': ''}],
         'meta': [{'name': 'Header', 'type': 'STRING_TYPE', 'comment': ''}],
         'type': 'table',
         'type': 'table',
         'has_more': False,
         'has_more': False,
     }
     }
 
 
+
   def cancel(self, notebook, snippet):
   def cancel(self, notebook, snippet):
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
 
 
@@ -99,13 +112,13 @@ class OozieApi(Api):
 
 
     return {'status': 0}
     return {'status': 0}
 
 
+
   def get_log(self, notebook, snippet, startFrom=0, size=None):
   def get_log(self, notebook, snippet, startFrom=0, size=None):
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
 
 
     oozie_job = check_job_access_permission(self.request, job_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):
   def progress(self, snippet, logs):
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
@@ -113,8 +126,44 @@ class OozieApi(Api):
     oozie_job = check_job_access_permission(self.request, job_id)
     oozie_job = check_job_access_permission(self.request, job_id)
     return oozie_job.get_progress(),
     return oozie_job.get_progress(),
 
 
+
   def close_statement(self, snippet):
   def close_statement(self, snippet):
     pass
     pass
 
 
+
   def close_session(self, session):
   def close_session(self, session):
     pass
     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 logging
 import json
 import json
+import re
+import time
 
 
 from django.core.urlresolvers import reverse
 from django.core.urlresolvers import reverse
 from django.http import QueryDict
 from django.http import QueryDict
@@ -31,6 +33,7 @@ LOG = logging.getLogger(__name__)
 try:
 try:
   from pig import api
   from pig import api
   from pig.models import PigScript2, get_workflow_output, hdfs_link
   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
   from oozie.views.dashboard import check_job_access_permission, check_job_edition_permission
 except Exception, e:
 except Exception, e:
   LOG.exception('Pig application is not enabled: %s' % e)
   LOG.exception('Pig application is not enabled: %s' % e)
@@ -38,12 +41,15 @@ except Exception, e:
 
 
 class PigApi(Api):
 class PigApi(Api):
 
 
+  RESULTS_PATTERN = "(?P<results>>>> Invoking Pig command line now >>>.+<<< Invocation of Pig command completed <<<)"
+
   def __init__(self, *args, **kwargs):
   def __init__(self, *args, **kwargs):
     Api.__init__(self, *args, **kwargs)
     Api.__init__(self, *args, **kwargs)
 
 
     self.fs = self.request.fs
     self.fs = self.request.fs
     self.jt = self.request.jt
     self.jt = self.request.jt
 
 
+
   def execute(self, notebook, snippet):
   def execute(self, notebook, snippet):
 
 
     attrs = {
     attrs = {
@@ -65,16 +71,18 @@ class PigApi(Api):
       'has_result_set': True,
       'has_result_set': True,
     }
     }
 
 
+
   def check_status(self, notebook, snippet):
   def check_status(self, notebook, snippet):
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
 
 
     oozie_workflow = check_job_access_permission(self.request, job_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 is_really_done and not oozie_workflow.is_running():
       if oozie_workflow.status in ('KILLED', 'FAILED'):
       if oozie_workflow.status in ('KILLED', 'FAILED'):
         raise QueryError(_('The script failed to run and was stopped'))
         raise QueryError(_('The script failed to run and was stopped'))
-      if logs:
+      if results:
         status = 'available'
         status = 'available'
       else:
       else:
         status = 'running' # Tricky case when the logs are being moved by YARN at job completion
         status = 'running' # Tricky case when the logs are being moved by YARN at job completion
@@ -87,30 +95,21 @@ class PigApi(Api):
         'status': status
         '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):
   def fetch_result(self, notebook, snippet, rows, start_over):
     job_id = snippet['result']['handle']['id']
     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 {
     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': ''}],
         'meta': [{'name': 'Header', 'type': 'STRING_TYPE', 'comment': ''}],
         'type': 'table',
         'type': 'table',
         'has_more': False,
         'has_more': False,
     }
     }
 
 
+
   def cancel(self, notebook, snippet):
   def cancel(self, notebook, snippet):
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
 
 
@@ -121,13 +120,14 @@ class PigApi(Api):
 
 
     return {'status': 0}
     return {'status': 0}
 
 
+
   def get_log(self, notebook, snippet, startFrom=0, size=None):
   def get_log(self, notebook, snippet, startFrom=0, size=None):
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
 
 
     oozie_workflow = check_job_access_permission(self.request, job_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):
   def progress(self, snippet, logs):
     job_id = snippet['result']['handle']['id']
     job_id = snippet['result']['handle']['id']
@@ -135,8 +135,42 @@ class PigApi(Api):
     oozie_workflow = check_job_access_permission(self.request, job_id)
     oozie_workflow = check_job_access_permission(self.request, job_id)
     return oozie_workflow.get_progress(),
     return oozie_workflow.get_progress(),
 
 
+
   def close_statement(self, snippet):
   def close_statement(self, snippet):
     pass
     pass
 
 
+
   def close_session(self, session):
   def close_session(self, session):
     pass
     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
+