浏览代码

[core] Resource Manager RM HA

The [[yarn_clusters]] cluster configs in Hue ini need to be filled up with
at least 2 configs to configure Hue for RM HA.
submit_to must be True to have the RM taken into consideration.

The initial RM being used is the one named [[[default]]].
When submitting to Oozie, RM configs must specify the same logical_name.
When browsing JobBrowser, Job Browser will switch to the active RM on the fly if needed.
Romain Rigaux 11 年之前
父节点
当前提交
378f789

+ 21 - 1
apps/jobbrowser/src/jobbrowser/api.py

@@ -30,7 +30,7 @@ import hadoop.yarn.node_manager_api as node_manager_api
 from jobbrowser.conf import SHARE_JOBS
 from jobbrowser.models import Job, JobLinkage, TaskList, Tracker
 from jobbrowser.yarn_models import Application, Job as YarnJob, Container
-from hadoop.cluster import get_next_ha_mrcluster
+from hadoop.cluster import get_next_ha_mrcluster, get_next_ha_yarncluster
 from desktop.lib.exceptions_renderable import PopupException
 
 
@@ -66,6 +66,24 @@ def jt_ha(funct):
   return wraps(funct)(decorate)
 
 
+def rm_ha(funct):
+  """
+  Support RM HA by trying other RM API.
+  """
+  def decorate(api, *args, **kwargs):
+    try:
+      return funct(api, *args, **kwargs)
+    except Exception, ex:
+      if 'Connection refused' in str(ex):
+        LOG.info('JobTracker not available, trying JT plugin HA: %s.' % ex)
+        rm_ha = get_next_ha_yarncluster()
+        if rm_ha is not None:
+          config, api.resource_manager_api = rm_ha
+          return funct(api, *args, **kwargs)
+      raise ex
+  return wraps(funct)(decorate)
+
+
 class JobBrowserApi(object):
 
   def paginate_task(self, task_list, pagenum):
@@ -195,6 +213,7 @@ class YarnApi(JobBrowserApi):
   def get_job_link(self, job_id):
     return self.get_job(job_id)
 
+  @rm_ha
   def get_jobs(self, user, **kwargs):
     state_filters = {'running': 'UNDEFINED', 'completed': 'SUCCEEDED', 'failed': 'FAILED', 'killed': 'KILLED', }
     filters = {}
@@ -228,6 +247,7 @@ class YarnApi(JobBrowserApi):
                   user.is_superuser or
                   job.user == user.username, jobs)
 
+  @rm_ha
   def get_job(self, jobid):
     try:
       # App id

+ 1 - 1
apps/jobbrowser/src/jobbrowser/templates/jobs.mako

@@ -137,7 +137,7 @@ ${ components.menubar() }
     $(document).ajaxError(function (event, jqxhr, settings, exception) {
       if (jqxhr.status == 500) {
         window.clearInterval(_runningInterval);
-        $(document).trigger("error", "${_('There was a problem communicating with the server. Refresh the page.')}");
+        $(document).trigger("error", "${_('There was a problem communicating with the server: ')}" + $.parseJSON(jqxhr.responseText).message);
       }
     });
 

+ 1 - 1
apps/jobbrowser/src/jobbrowser/tests.py

@@ -60,7 +60,7 @@ class TestBrowser():
     assert_equal("Foo Bar Baz", views.format_counter_name("fooBarBaz"))
     assert_equal("Foo", views.format_counter_name("foo"))
     assert_equal("Foo.", views.format_counter_name("foo."))
-    assert_equal("A Bbb Ccc", views.format_counter_name("A_BBB_CCC"))\
+    assert_equal("A Bbb Ccc", views.format_counter_name("A_BBB_CCC"))
 
 def get_hadoop_job_id(oozie_api, oozie_jobid, action_index=1, timeout=60, step=5):
   hadoop_job_id = None

+ 14 - 3
desktop/conf.dist/hue.ini

@@ -584,6 +584,9 @@
       # Whether to submit jobs to this cluster
       submit_to=True
 
+      # Resource Manager logical name (required for HA)
+      ## logical_name=
+
       # Change this if your YARN cluster is Kerberos-secured
       ## security_enabled=false
 
@@ -596,6 +599,13 @@
       # URL of the HistoryServer API
       # history_server_api_url=http://localhost:19888
 
+    # HA support by specifying multiple clusters
+    # e.g.
+
+    # [[[ha]]]
+      # Resource Manager logical name (required for HA)
+      # logical_name=my-rm-name
+
   # Configuration for MapReduce (MR1)
   # ------------------------------------------------------------------------
   [[mapred_clusters]]
@@ -607,7 +617,7 @@
       # The port where the JobTracker IPC listens on
       ## jobtracker_port=8021
 
-      # JobTracker logical name.
+      # JobTracker logical name for HA
       ## logical_name=
 
       # Thrift plug-in port for the JobTracker
@@ -623,8 +633,9 @@
     # e.g.
 
     # [[[ha]]]
-      # Enter the host on which you are running the failover JobTracker
-      # jobtracker_host=localhost-ha
+      # Enter the logical name of the JobTrackers
+      # logical_name=my-jt-name
+
 
 ###########################################################################
 # Settings to configure the Filebrowser app

+ 15 - 40
desktop/conf/pseudo-distributed.ini.tmpl

@@ -418,7 +418,7 @@
 
 
 ###########################################################################
-# Settings to configure OPENID
+# Settings to configure OpenId
 ###########################################################################
 
 [libopenid]
@@ -577,18 +577,6 @@
       # Change this if your HDFS cluster is Kerberos-secured
       ## security_enabled=false
 
-      # Settings about this HDFS cluster. If you install HDFS in a
-      # different location, you need to set the following.
-
-      # Defaults to $HADOOP_HDFS_HOME or /usr/lib/hadoop-hdfs
-      ## hadoop_hdfs_home=/usr/lib/hadoop-hdfs
-
-      # Defaults to $HADOOP_BIN or /usr/bin/hadoop
-      ## hadoop_bin=/usr/bin/hadoop
-
-      # Defaults to $HADOOP_CONF_DIR or /etc/hadoop/conf
-      ## hadoop_conf_dir=/etc/hadoop/conf
-
   # Configuration for YARN (MR2)
   # ------------------------------------------------------------------------
   [[yarn_clusters]]
@@ -603,21 +591,12 @@
       # Whether to submit jobs to this cluster
       submit_to=True
 
+      # Resource Manager logical name (required for HA)
+      ## logical_name=
+
       # Change this if your YARN cluster is Kerberos-secured
       ## security_enabled=false
 
-      # Settings about this MR2 cluster. If you install MR2 in a
-      # different location, you need to set the following.
-
-      # Defaults to $HADOOP_MR2_HOME or /usr/lib/hadoop-mapreduce
-      ## hadoop_mapred_home=/usr/lib/hadoop-mapreduce
-
-      # Defaults to $HADOOP_BIN or /usr/bin/hadoop
-      ## hadoop_bin=/usr/bin/hadoop
-
-      # Defaults to $HADOOP_CONF_DIR or /etc/hadoop/conf
-      ## hadoop_conf_dir=/etc/hadoop/conf
-
       # URL of the ResourceManager API
       ## resourcemanager_api_url=http://localhost:8088
 
@@ -627,6 +606,13 @@
       # URL of the HistoryServer API
       # history_server_api_url=http://localhost:19888
 
+    # HA support by specifying multiple clusters
+    # e.g.
+
+    # [[[ha]]]
+      # Resource Manager logical name (required for HA)
+      ## logical_name=my-rm-name
+
   # Configuration for MapReduce (MR1)
   # ------------------------------------------------------------------------
   [[mapred_clusters]]
@@ -638,7 +624,7 @@
       # The port where the JobTracker IPC listens on
       ## jobtracker_port=8021
 
-      # JobTracker logical name.
+      # JobTracker logical name for HA
       ## logical_name=
 
       # Thrift plug-in port for the JobTracker
@@ -650,24 +636,13 @@
       # Change this if your MapReduce cluster is Kerberos-secured
       ## security_enabled=false
 
-      # Settings about this MR1 cluster. If you install MR1 in a
-      # different location, you need to set the following.
-
-      # Defaults to $HADOOP_MR1_HOME or /usr/lib/hadoop-0.20-mapreduce
-      ## hadoop_mapred_home=/usr/lib/hadoop-0.20-mapreduce
-
-      # Defaults to $HADOOP_BIN or /usr/bin/hadoop
-      ## hadoop_bin=/usr/bin/hadoop
-
-      # Defaults to $HADOOP_CONF_DIR or /etc/hadoop/conf
-      ## hadoop_conf_dir=/etc/hadoop/conf
-
     # HA support by specifying multiple clusters
     # e.g.
 
     # [[[ha]]]
-      # Enter the host on which you are running the failover JobTracker
-      # jobtracker_host=localhost-ha
+      # Enter the logical name of the JobTrackers
+      # logical_name=my-jt-name
+
 
 ###########################################################################
 # Settings to configure the Filebrowser app

+ 1 - 1
desktop/core/src/desktop/middleware.py

@@ -145,7 +145,7 @@ class ClusterMiddleware(object):
       if request.fs is not None:
         request.fs.setuser(request.user.username)
 
-      request.jt = cluster.get_default_mrcluster()
+      request.jt = cluster.get_default_mrcluster() # Deprecated, only there for MR1
       if request.jt is not None:
         request.jt.setuser(request.user.username)
     else:

+ 87 - 22
desktop/libs/hadoop/src/hadoop/cluster.py

@@ -15,13 +15,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from hadoop.fs import hadoopfs, webhdfs, LocalSubFileSystem
+import os
+import logging
+
+from hadoop import conf
+from hadoop.fs import webhdfs, LocalSubFileSystem
 from hadoop.job_tracker import LiveJobTracker
 
 from desktop.lib.paths import get_build_dir
-from hadoop import conf
-import os
-import logging
+
 
 LOG = logging.getLogger(__name__)
 
@@ -33,32 +35,28 @@ MR_NAME_CACHE = 'default'
 
 def _make_filesystem(identifier):
   choice = os.getenv("FB_FS")
+
   if choice == "testing":
     path = os.path.join(get_build_dir(), "fs")
     if not os.path.isdir(path):
-      LOG.warning(
-        ("Could not find fs directory: %s. Perhaps you need to run " +
-        "manage.py filebrowser_test_setup?") % path)
+      LOG.warning(("Could not find fs directory: %s. Perhaps you need to run manage.py filebrowser_test_setup?") % path)
     return LocalSubFileSystem(path)
   else:
     cluster_conf = conf.HDFS_CLUSTERS[identifier]
-    # The only way to disable webhdfs is to specify an empty value
-    if cluster_conf.WEBHDFS_URL.get() != '':
-      return webhdfs.WebHdfs.from_config(cluster_conf)
-    else:
-      return hadoopfs.HadoopFileSystem.from_config(
-        cluster_conf,
-        hadoop_bin_path=conf.HADOOP_BIN.get())
+    return webhdfs.WebHdfs.from_config(cluster_conf)
+
 
 def _make_mrcluster(identifier):
   cluster_conf = conf.MR_CLUSTERS[identifier]
   return LiveJobTracker.from_conf(cluster_conf)
 
+
 def get_hdfs(identifier="default"):
   global FS_CACHE
   get_all_hdfs()
   return FS_CACHE[identifier]
 
+
 def get_all_hdfs():
   global FS_CACHE
   if FS_CACHE is not None:
@@ -69,6 +67,7 @@ def get_all_hdfs():
     FS_CACHE[identifier] = _make_filesystem(identifier)
   return FS_CACHE
 
+
 def get_default_mrcluster():
   """
   Get the default JT (not necessarily HA).
@@ -86,6 +85,19 @@ def get_default_mrcluster():
       return candidates.values()[0]
     return None
 
+
+def get_default_yarncluster():
+  """
+  Get the default RM (not necessarily HA).
+  """
+  global MR_NAME_CACHE
+
+  try:
+    return conf.YARN_CLUSTERS[MR_NAME_CACHE]
+  except KeyError:
+    return get_yarn()
+
+
 def get_next_ha_mrcluster():
   """
   Return the next available JT instance and cache its name.
@@ -120,11 +132,13 @@ def get_next_ha_mrcluster():
         return (config, jt)
   return None
 
+
 def get_mrcluster(identifier="default"):
   global MR_CACHE
   all_mrclusters()
   return MR_CACHE[identifier]
 
+
 def all_mrclusters():
   global MR_CACHE
   if MR_CACHE is not None:
@@ -134,44 +148,94 @@ def all_mrclusters():
     MR_CACHE[identifier] = _make_mrcluster(identifier)
   return MR_CACHE
 
+
 def get_yarn():
+  global MR_NAME_CACHE
+  if MR_NAME_CACHE in conf.YARN_CLUSTERS:
+    return conf.YARN_CLUSTERS[MR_NAME_CACHE]
+
   for name in conf.YARN_CLUSTERS.keys():
     yarn = conf.YARN_CLUSTERS[name]
     if yarn.SUBMIT_TO.get():
       return yarn
 
-def get_cluster_conf_for_job_submission():
+
+def get_next_ha_yarncluster():
   """
-  Check the `submit_to' for each MR/Yarn cluster, and return the
+  Return the next available YARN RM instance and cache its name.
+  """
+  from hadoop.yarn.resource_manager_api import ResourceManagerApi
+  global MR_NAME_CACHE
+
+  has_ha = sum([conf.YARN_CLUSTERS[name].SUBMIT_TO.get() for name in conf.YARN_CLUSTERS.keys()]) >= 2
+
+  for name in conf.YARN_CLUSTERS.keys():
+    config = conf.YARN_CLUSTERS[name]
+    if config.SUBMIT_TO.get():
+      if has_ha:
+        try:
+          rm = ResourceManagerApi(config.RESOURCE_MANAGER_API_URL.get(), config.SECURITY_ENABLED.get())
+          cluster_info = rm.cluster()
+          if cluster_info['clusterInfo']['haState'] == 'ACTIVE':
+            MR_NAME_CACHE = name
+            LOG.warn('Picking RM HA: %s' % name)
+            return (config, rm)
+          else:
+            LOG.info('RM %s is not RUNNING, skipping it: %s' % (name, cluster_info))
+        except Exception, ex:
+          LOG.info('RM %s is not available, skipping it: %s' % (name, ex))
+      else:
+        return (config, rm)
+  return None
+
+
+def get_cluster_for_job_submission():
+  """
+  Check the 'submit_to' for each MR/Yarn cluster, and return the
   config section of first one that enables submission.
 
-  HA support for MR1.
+  Support MR1/MR2 HA.
   """
-  yarn = get_yarn()
+  yarn = get_next_ha_yarncluster()
   if yarn:
     return yarn
 
   mr = get_next_ha_mrcluster()
-
   if mr is not None:
-    config, jt = mr
+    return mr
+
+  return None
+
+
+def get_cluster_conf_for_job_submission():
+  cluster = get_cluster_for_job_submission()
+
+  if cluster:
+    config, rm = cluster
     return config
   else:
     return None
 
+
 def get_cluster_addr_for_job_submission():
   """
-  Check the `submit_to' for each MR/Yarn cluster, and return the
-  host:port of first one that enables submission.
+  Check the 'submit_to' for each MR/Yarn cluster, and return the logical name or host:port of first one that enables submission.
   """
+  if is_yarn():
+    if get_yarn().LOGICAL_NAME.get():
+      return get_yarn().LOGICAL_NAME.get()
+
   conf = get_cluster_conf_for_job_submission()
   if conf is None:
     return None
+
   return "%s:%s" % (conf.HOST.get(), conf.PORT.get())
 
+
 def is_yarn():
   return get_yarn() is not None
 
+
 def clear_caches():
   """
   Clears cluster's internal caches.  Returns
@@ -182,6 +246,7 @@ def clear_caches():
   FS_CACHE, MR_CACHE = None, None
   return old
 
+
 def restore_caches(old):
   """
   Restores caches from the result of a previous clear_caches call.

+ 9 - 6
desktop/libs/hadoop/src/hadoop/conf.py

@@ -84,10 +84,10 @@ MR_CLUSTERS = UnspecifiedConfigSection(
                   default=8021,
                   help="Service port for the JobTracker",
                   type=int),
-      LOGICAL_NAME = Config('logical_name',
-                            default="",
-                            type=str,
-                            help=_t('JobTracker logical name.')),
+      LOGICAL_NAME=Config('logical_name',
+                          default="",
+                          type=str,
+                          help=_t('JobTracker logical name.')),
       JT_THRIFT_PORT=Config("thrift_port", help="Thrift port for JobTracker", default=9290,
                             type=int),
       JT_KERBEROS_PRINCIPAL=Config("jt_kerberos_principal", help="Kerberos principal for JobTracker",
@@ -102,8 +102,7 @@ MR_CLUSTERS = UnspecifiedConfigSection(
 
 YARN_CLUSTERS = UnspecifiedConfigSection(
   "yarn_clusters",
-  help="One entry for each Yarn cluster. Currently only one cluster "
-       "(called 'default') is supported.",
+  help="One entry for each Yarn cluster",
   each=ConfigSection(
     help="Information about a single Yarn cluster",
     members=dict(
@@ -114,6 +113,10 @@ YARN_CLUSTERS = UnspecifiedConfigSection(
                   default=8032,
                   type=int,
                   help="Service port for the ResourceManager"),
+      LOGICAL_NAME=Config('logical_name',
+                          default="",
+                          type=str,
+                          help=_t('Resource Manager logical name.')),
       SECURITY_ENABLED=Config("security_enabled", help="Is running with Kerberos authentication",
                               default=False, type=coerce_bool),
       SUBMIT_TO=Config('submit_to', help="Whether Hue should use this cluster to run jobs",

+ 9 - 0
desktop/libs/hadoop/src/hadoop/yarn/resource_manager_api.py

@@ -19,8 +19,12 @@ import logging
 import posixpath
 import threading
 
+from django.utils.translation import ugettext as _
+
+from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.rest.http_client import HttpClient
 from desktop.lib.rest.resource import Resource
+
 from hadoop import cluster
 
 
@@ -41,6 +45,8 @@ def get_resource_manager():
     try:
       if _api_cache is None:
         yarn_cluster = cluster.get_cluster_conf_for_job_submission()
+        if yarn_cluster is None:
+          raise PopupException(_('No Resource Manager are available.'))
         _api_cache = ResourceManagerApi(yarn_cluster.RESOURCE_MANAGER_API_URL.get(), yarn_cluster.SECURITY_ENABLED.get())
     finally:
       _api_cache_lock.release()
@@ -68,6 +74,9 @@ class ResourceManagerApi(object):
   def security_enabled(self):
     return self._security_enabled
 
+  def cluster(self, **kwargs):
+    return self._root.get('cluster', params=kwargs, headers={'Accept': _JSON_CONTENT_TYPE})
+
   def apps(self, **kwargs):
     return self._root.get('cluster/apps', params=kwargs, headers={'Accept': _JSON_CONTENT_TYPE})
 

+ 11 - 10
desktop/libs/liboozie/src/liboozie/submittion.py

@@ -47,7 +47,7 @@ class Submission(object):
     self.job = job
     self.user = user
     self.fs = fs
-    self.jt = jt
+    self.jt = jt # Deprecated with YARN, we now use logical names only for RM
     self.oozie_id = oozie_id
     self.api = get_oozie(self.user)
 
@@ -73,13 +73,13 @@ class Submission(object):
     if self.oozie_id is not None:
       raise Exception(_("Submission already submitted (Oozie job id %s)") % (self.oozie_id,))
 
-    jobtracker = cluster.get_cluster_addr_for_job_submission()
+    jt_address = cluster.get_cluster_addr_for_job_submission()
 
     if deployment_dir is None:
-      self._update_properties(jobtracker) # Needed as we need to set some properties like Credentials before
+      self._update_properties(jt_address) # Needed as we need to set some properties like Credentials before
       deployment_dir = self.deploy()
 
-    self._update_properties(jobtracker, deployment_dir)
+    self._update_properties(jt_address, deployment_dir)
 
     self.oozie_id = self.api.submit_job(self.properties)
     LOG.info("Submitted: %s" % (self,))
@@ -91,9 +91,9 @@ class Submission(object):
     return self.oozie_id
 
   def rerun(self, deployment_dir, fail_nodes=None, skip_nodes=None):
-    jobtracker = cluster.get_cluster_addr_for_job_submission()
+    jt_address = cluster.get_cluster_addr_for_job_submission()
 
-    self._update_properties(jobtracker, deployment_dir)
+    self._update_properties(jt_address, deployment_dir)
     self.properties.update({'oozie.wf.application.path': deployment_dir})
 
     if fail_nodes:
@@ -111,9 +111,9 @@ class Submission(object):
 
 
   def rerun_coord(self, deployment_dir, params):
-    jobtracker = cluster.get_cluster_addr_for_job_submission()
+    jt_address = cluster.get_cluster_addr_for_job_submission()
 
-    self._update_properties(jobtracker, deployment_dir)
+    self._update_properties(jt_address, deployment_dir)
     self.properties.update({'oozie.coord.application.path': deployment_dir})
 
     self.api.job_control(self.oozie_id, action='coord-rerun', properties=self.properties, parameters=params)
@@ -123,9 +123,9 @@ class Submission(object):
 
 
   def rerun_bundle(self, deployment_dir, params):
-    jobtracker = cluster.get_cluster_addr_for_job_submission()
+    jt_address = cluster.get_cluster_addr_for_job_submission()
 
-    self._update_properties(jobtracker, deployment_dir)
+    self._update_properties(jt_address, deployment_dir)
     self.properties.update({'oozie.bundle.application.path': deployment_dir})
     self.api.job_control(self.oozie_id, action='bundle-rerun', properties=self.properties, parameters=params)
     LOG.info("Rerun: %s" % (self,))
@@ -180,6 +180,7 @@ class Submission(object):
 
   def _update_properties(self, jobtracker_addr, deployment_dir=None):
     LOG.info('Using FS %s and JT %s' % (self.fs, self.jt))
+
     if self.jt and self.jt.logical_name:
       jobtracker_addr = self.jt.logical_name