소스 검색

HUE-7803 [core] Remove deprecated JobTracker Python class

Romain Rigaux 8 년 전
부모
커밋
70d2f57
2개의 변경된 파일3개의 추가작업 그리고 479개의 파일을 삭제
  1. 3 74
      desktop/libs/hadoop/src/hadoop/fs/hadoopfs.py
  2. 0 405
      desktop/libs/hadoop/src/hadoop/job_tracker.py

+ 3 - 74
desktop/libs/hadoop/src/hadoop/fs/hadoopfs.py

@@ -30,13 +30,11 @@ import random
 import subprocess
 import urlparse
 
-from thrift.transport import TTransport
-
 from django.utils.encoding import smart_str, force_unicode
 from django.utils.translation import ugettext as _
+
 from desktop.lib import i18n
-from desktop.lib.conf import validate_port
-from hadoop.api.common.ttypes import RequestContext, IOException
+
 import hadoop.conf
 from hadoop.fs import normpath, SEEK_SET, SEEK_CUR, SEEK_END
 from hadoop.fs.exceptions import PermissionDeniedException
@@ -71,75 +69,6 @@ def decode_fs_path(path):
   return force_unicode(path, HDFS_ENCODING, errors='strict')
 
 
-def test_fs_configuration(fs_config, hadoop_bin_conf):
-  """Test FS configuration. Returns list of (confvar, error)."""
-  TEST_FILE = '/tmp/.hue_config_test.%s' % (random.randint(0, 9999999999))
-  res = [ ]
-
-  res.extend(validate_port(fs_config.NN_THRIFT_PORT))
-  res.extend(validate_port(fs_config.NN_HDFS_PORT))
-  if res:
-    return res
-
-  # Check thrift plugin
-  try:
-    fs = HadoopFileSystem.from_config(
-      fs_config, hadoop_bin_path=hadoop_bin_conf.get())
-
-    fs.setuser(fs.superuser)
-    ls = fs.listdir('/')
-  except TTransport.TTransportException:
-    msg = 'Failed to contact Namenode plugin at %s:%s.' % \
-            (fs_config.NN_HOST.get(), fs_config.NN_THRIFT_PORT.get())
-    LOG.exception(msg)
-    res.append((fs_config, msg))
-    return res
-  except (IOError, IOException):
-    msg = 'Failed to see HDFS root directory at %s. Please check HDFS configuration.' % (fs.uri,)
-    LOG.exception(msg)
-    res.append((fs_config, msg))
-    return res
-
-  if 'tmp' not in ls:
-    return res
-
-  # Check nn port (via upload)
-  try:
-    w_file = fs.open(TEST_FILE, 'w')
-  except OSError, ex:
-    msg = 'Failed to execute Hadoop (%s)' % (hadoop_bin_conf.get(),)
-    LOG.exception(msg)
-    res.append((hadoop_bin_conf, msg))
-    return res
-
-  try:
-    try:
-      w_file.write('hello world')
-      w_file.close()
-    except IOError:
-      msg = 'Failed to upload files using %s' % (fs.uri,)
-      LOG.exception(msg)
-      res.append((fs_config.NN_HDFS_PORT, msg))
-      return res
-
-    # Check dn plugin (via read)
-    try:
-      r_file = fs.open(TEST_FILE, 'r')
-      r_file.read()
-    except Exception:
-      msg = 'Failed to read file. Are all datanodes configured with the HUE plugin?'
-      LOG.exception(msg)
-      res.append((fs_config, msg))
-  finally:
-    # Cleanup. Ignore if file not found.
-    try:
-      if fs.exists(TEST_FILE):
-        fs.remove(TEST_FILE)
-    except Exception, ex:
-      LOG.error('Failed to cleanup test file "%s:%s": %s' % (fs.uri, TEST_FILE, ex))
-  return res
-
-
 def _coerce_exceptions(function):
   """
   Decorator that causes exceptions thrown by the decorated function
@@ -149,7 +78,7 @@ def _coerce_exceptions(function):
   def wrapper(*args, **kwargs):
     try:
       return function(*args, **kwargs)
-    except IOException, e:
+    except Exception, e:
       e.msg = force_unicode(e.msg, errors='replace')
       e.stack = force_unicode(e.stack, errors='replace')
       LOG.exception("Exception in Hadoop FS call " + function.__name__)

+ 0 - 405
desktop/libs/hadoop/src/hadoop/job_tracker.py

@@ -1,405 +0,0 @@
-#!/usr/bin/env python
-# Licensed to Cloudera, Inc. under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  Cloudera, Inc. licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-# Django-side implementation of the JobTracker plugin interface
-
-from desktop.lib import thrift_util
-from desktop.lib.conf import validate_port
-from desktop.lib.exceptions import StructuredException
-from desktop.lib.thrift_util import fixup_enums
-
-from hadoop.api.jobtracker import Jobtracker
-from hadoop.api.jobtracker.ttypes import ThriftJobID, ThriftTaskAttemptID, \
-    ThriftTaskType, ThriftTaskPhase, ThriftTaskID, \
-    ThriftTaskState, ThriftJobState, ThriftJobPriority, TaskNotFoundException, \
-    JobTrackerState, JobNotFoundException, ThriftTaskQueryState
-from hadoop.api.common.ttypes import RequestContext
-from thrift.transport import TTransport
-
-import threading
-
-VALID_TASK_STATES = set(["succeeded", "failed", "running", "pending", "killed"])
-VALID_TASK_TYPES = set(["map", "reduce", "job_cleanup", "job_setup"])
-
-# timeout (seconds) for thrift calls to jobtracker
-JT_THRIFT_TIMEOUT = 15
-
-DEFAULT_USER = "webui"
-
-def test_jt_configuration(cluster):
-  """Test FS configuration. Returns list of (confvar, error)."""
-  err = validate_port(cluster.JT_THRIFT_PORT)
-  if err:
-    return err
-
-  try:
-    jt = LiveJobTracker.from_conf(cluster)
-    jt.runtime_info()
-  except TTransport.TTransportException:
-    msg = 'Failed to contact JobTracker plugin at %s:%s.' % \
-          (cluster.HOST.get(), cluster.JT_THRIFT_PORT.get())
-    return [ (cluster, msg) ]
-  return []
-
-
-class LiveJobTracker(object):
-  """
-  Connects to a JobTracker over our Thrift interface.
-  At the moment, doesn't handle Thrift errors gracefully.
-
-  In particular, if Thrift returns None for anything, this will throw.
-  """
-
-  def __init__(self, host,
-               thrift_port,
-               logical_name=None,
-               security_enabled=False,
-               kerberos_principal="mapred"):
-    self.client = thrift_util.get_client(
-      Jobtracker.Client, host, thrift_port,
-      service_name="Hadoop MR JobTracker HUE Plugin",
-      use_sasl=security_enabled,
-      kerberos_principal=kerberos_principal,
-      timeout_seconds=JT_THRIFT_TIMEOUT)
-    self.host = host
-    self.thrift_port = thrift_port
-    self.logical_name = logical_name
-    self.security_enabled = security_enabled
-    # We allow a single LiveJobTracker to be used across multiple
-    # threads by restricting the stateful components to a thread
-    # thread-local.
-    self.thread_local = threading.local()
-    self.setuser(DEFAULT_USER)
-
-  @classmethod
-  def from_conf(cls, conf):
-    return cls(
-      conf.HOST.get(),
-      conf.JT_THRIFT_PORT.get(),
-      conf.LOGICAL_NAME.get(),
-      security_enabled=conf.SECURITY_ENABLED.get(),
-      kerberos_principal=conf.JT_KERBEROS_PRINCIPAL.get())
-
-  def thriftjobid_from_string(self, jobid):
-    """The jobid looks like this: job_201001301455_0001"""
-    _, tid, jid = jobid.split("_")
-    return ThriftJobID(tid, int(jid), jobid)
-
-  def thriftattemptid_from_string(self, attemptid):
-    """The attemptid looks like this: attempt_201001301455_0001_m_000007_0"""
-    _, meat = attemptid.split('_', maxsplit=1)
-    taskid, aid = meat.rsplit('_', maxsplit=1)
-    return ThriftTaskAttemptID(self.thrifttaskid_from_string(taskid), int(aid))
-
-  def thrifttaskid_from_string(self, taskid_str):
-    """The taskid_str looks like this: task_201001301455_0001_m_000007"""
-    TASK_TYPE_MAP = { "m" : ThriftTaskType.MAP,
-                      "r" : ThriftTaskType.REDUCE }
-    _, tracker, jid, tasktype, tid = taskid_str.split('_')
-    return ThriftTaskID(ThriftJobID(tracker, int(jid)),
-                        TASK_TYPE_MAP[tasktype],
-                        int(tid))
-
-  def _fixup_tasktracker(self, tracker):
-    """
-    Adds string representations of enums to a ThirftTaskTrackerStatus
-    """
-    if not tracker.taskReports:
-      return tracker
-    for tts in tracker.taskReports:
-      fixup_enums(tts.taskID.taskID, {"taskType":ThriftTaskType})
-      self._fixup_taskstatus(tts)
-
-
-  def _fixup_taskstatus(self, taskstatus):
-    """
-    Adds string representations of enums to a ThriftTaskInProgress
-    """
-    fixup_enums(taskstatus, {"state":ThriftTaskState, "phase":ThriftTaskPhase})
-    fixup_enums(taskstatus.taskID.taskID, {"taskType":ThriftTaskType})
-
-  def _fixup_task_state(self, tip):
-    """
-    Determine overall state of a task.
-    """
-    if tip.complete:
-      tip.state = "succeeded"
-    elif tip.failed:
-      if not tip.execStartTime:
-        tip.state = "killed"
-      else:
-        tip.state = "failed"
-    elif tip.execStartTime:
-      tip.state = "running"
-    else:
-      tip.state = "pending"
-
-    global VALID_TASK_STATES
-    assert tip.state in VALID_TASK_STATES
-
-  def _fixup_job(self, job):
-    """
-    Run through all the enums in a job and add their string variant
-    """
-    fixup_enums(job.status, {"runState":ThriftJobState, "priority":ThriftJobPriority})
-    fixup_enums(job, {"priority":ThriftJobPriority})
-    if job.tasks is not None:
-      for tip in job.tasks.tasks:
-        fixup_enums(tip.taskID, {"taskType":ThriftTaskType})
-        self._fixup_task_state(tip)
-        for taskstatus in tip.taskStatuses.values():
-          self._fixup_taskstatus(taskstatus)
-
-  def _fixup_task_in_progress(self, tip):
-    fixup_enums(tip.taskID, {"taskType": ThriftTaskType})
-    self._fixup_task_state(tip)
-    for taskstatus in tip.taskStatuses.values():
-      self._fixup_taskstatus(taskstatus)
-
-  def _fixup_retired_job(self, job):
-    job.is_retired = True
-
-  def setuser(self, user):
-    # Hadoop determines the groups the user belongs to on the server side.
-    self.thread_local.request_context = RequestContext()
-    if not self.thread_local.request_context.confOptions:
-      self.thread_local.request_context.confOptions = {}
-    self.thread_local.request_context.confOptions['effective_user'] = user
-    self.thread_local.user = user
-
-  @property
-  def user(self):
-    return self.thread_local.user
-
-  @property
-  def request_context(self):
-    # Here for backwards-compatibility.
-    return self.thread_local.request_context
-
-
-  def queues(self):
-    """
-    Returns a ThriftJobQueueList
-    """
-    qs = self.client.getQueues(self.thread_local.request_context)
-    return qs
-
-  def cluster_status(self):
-    """
-    Returns a ThriftClusterStatus
-    """
-    cs = self.client.getClusterStatus(self.thread_local.request_context)
-    fixup_enums(cs, {"state":JobTrackerState})
-    return cs
-
-  def runtime_info(self):
-    """
-    Returns a RuntimeInfo
-    """
-    return self.client.getRuntimeInfo(self.thread_local.request_context)
-
-  def all_task_trackers(self):
-    """
-    Returns a ThriftTaskTrackerStatusList
-    """
-    tts = self.client.getAllTrackers(self.thread_local.request_context)
-    for tracker in tts.trackers:
-      self._fixup_tasktracker(tracker)
-    return tts
-
-  def active_trackers(self):
-    """
-    Returns a ThriftTaskTrackerStatusList
-    """
-    tts = self.client.getActiveTrackers(self.thread_local.request_context)
-    for tracker in tts.trackers:
-      self._fixup_tasktracker(tracker)
-    return tts
-
-  def blacklisted_trackers(self):
-    """
-    Returns a ThriftTaskTrackerStatusList
-    """
-    tts = self.client.getBlacklistedTrackers(self.thread_local.request_context)
-    for tracker in tts.trackers:
-      self._fixup_tasktracker(tracker)
-    return tts
-
-  def task_tracker(self, name):
-    """
-    Returns a ThriftTaskTrackerStatus or None
-    """
-    tracker = self.client.getTracker(self.thread_local.request_context, name)
-    if not tracker:
-      return None
-    self._fixup_tasktracker(tracker)
-    return tracker
-
-  def get_job(self, jobid):
-    """
-    Returns a ThriftJobInProgress (including task info)
-    """
-    try:
-      job = self.client.getJob(self.thread_local.request_context, jobid)
-    except JobNotFoundException, e:
-      raise StructuredException(code="JT_JOB_NOT_FOUND", message="Could not find job %s on JobTracker." % jobid.asString, data=jobid)
-    self._fixup_job(job)
-    return job
-
-  def running_jobs(self):
-    """
-    Returns a ThriftJobList (does not include task info)
-    """
-    joblist = self.client.getRunningJobs(self.thread_local.request_context)
-    for job in joblist.jobs:
-      self._fixup_job(job)
-    return joblist
-
-  def completed_jobs(self):
-    """
-    Returns a ThriftJobList (does not include task info)
-    """
-    joblist = self.client.getCompletedJobs(self.thread_local.request_context)
-    for job in joblist.jobs:
-      self._fixup_job(job)
-    return joblist
-
-  def get_retired_job(self, jobid):
-    """
-    Returns a ThriftJobInProgress (does not include task info and most of the job information)
-    """
-    try:
-      job = self.client.getRetiredJob(self.thread_local.request_context, jobid)
-    except JobNotFoundException, e:
-        raise StructuredException(code="JT_JOB_NOT_FOUND", message="Could not find job %s on JobTracker." % jobid.asString, data=jobid)
-    self._fixup_job(job)
-    self._fixup_retired_job(job)
-    return job
-
-  def retired_jobs(self, status=None):
-    """
-    Returns a ThriftJobStatusList (does not include task info and most of the job information)
-    """
-    joblist = self.client.getRetiredJobs(self.thread_local.request_context, status)
-    for job in joblist.jobs:
-      self._fixup_job(job)
-      self._fixup_retired_job(job)
-    return joblist
-
-  def failed_jobs(self):
-    """
-    Returns a ThriftJobList (does not include task info)
-    """
-    joblist = self.client.getFailedJobs(self.thread_local.request_context)
-    for job in joblist.jobs:
-      self._fixup_job(job)
-    return joblist
-
-  def killed_jobs(self):
-    """
-    Returns a ThriftJobList (does not include task info)
-    """
-    joblist = self.client.getKilledJobs(self.thread_local.request_context)
-    for job in joblist.jobs:
-      self._fixup_job(job)
-    return joblist
-
-  def all_jobs(self):
-    """
-    Returns a ThriftJobList (does not include task info)
-    """
-    joblist = self.client.getAllJobs(self.thread_local.request_context)
-    for job in joblist.jobs:
-      self._fixup_job(job)
-    return joblist
-
-  def get_job_count_by_user(self, user):
-    """
-    Returns a ThriftUserJobCounts.
-    """
-    return self.client.getUserJobCounts(self.thread_local.request_context, user)
-
-  def get_job_counters(self, jobid):
-    """
-    Returns a ThriftGroupList
-    """
-    return self.client.getJobCounters(self.thread_local.request_context, jobid)
-
-  def get_job_counter_rollups(self, jobid):
-    """
-    Returns a ThriftGroupList
-    """
-    return self.client.getJobCounterRollups(self.thread_local.request_context, jobid)
-
-
-  def get_task_list(self, jobid, task_types, task_states, task_text, count, offset):
-    """
-    Return a ThriftTaskInProgressList
-    """
-    # Translate task_states and task_types to their thrift variants
-    ttask_types = [ ThriftTaskType._NAMES_TO_VALUES[x.upper()] for x in task_types ]
-    ttask_states = [ ThriftTaskQueryState._NAMES_TO_VALUES[x.upper()] for x in task_states ]
-    tip_list = self.client.getTaskList(
-          self.thread_local.request_context, jobid, ttask_types, ttask_states, task_text, count, offset)
-
-    for tip in tip_list.tasks:
-      self._fixup_task_in_progress(tip)
-    return tip_list
-
-  def get_task(self, jobid, taskid):
-    """Return a ThriftTaskInProgress"""
-    try:
-      tip = self.client.getTask(self.thread_local.request_context, taskid)
-    except JobNotFoundException, e:
-      raise StructuredException(code="JT_JOB_NOT_FOUND", message="Could not find job %s on JobTracker." % jobid.asString, data=jobid)
-    except TaskNotFoundException, e:
-      raise StructuredException(code="JT_TASK_NOT_FOUND", message="Could not find task %s on JobTracker." % taskid.asString, data=taskid)
-    self._fixup_task_in_progress(tip)
-    return tip
-
-  def get_current_time(self):
-    """
-    Returns an integer timestamp
-    """
-    return self.client.getCurrentTime(self.thread_local.request_context)
-
-  def get_job_xml(self, jobid):
-    """
-    Returns a string representation of the job XML
-    """
-    return self.client.getJobConfXML(self.thread_local.request_context, jobid)
-
-  def kill_job(self, jobid):
-    """
-    Kill a job
-    """
-    return self.client.killJob(self.thread_local.request_context, jobid)
-
-  def kill_task_attempt(self, attemptid):
-    """
-    Kill a task attempt
-    """
-    return self.client.killTaskAttempt(self.thread_local.request_context, attemptid)
-
-  def set_job_priority(self, jobid, priority):
-    """
-    Set a job's priority
-    """
-    return self.client.setJobPriority(self.thread_local.request_context, jobid, priority)
-
-  def get_delegation_token(self, principal):
-    return self.client.getDelegationToken(self.thread_local.request_context, principal).delegationTokenBytes