فهرست منبع

[core] Support Resource Manager HA active / standby switching

Used to only failover when the RM was going down.
Invalidate the current RM API when switching RM.
Romain Rigaux 11 سال پیش
والد
کامیت
0109c9d
2فایلهای تغییر یافته به همراه9 افزوده شده و 3 حذف شده
  1. 6 2
      apps/jobbrowser/src/jobbrowser/api.py
  2. 3 1
      desktop/libs/hadoop/src/hadoop/cluster.py

+ 6 - 2
apps/jobbrowser/src/jobbrowser/api.py

@@ -74,8 +74,9 @@ def rm_ha(funct):
     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)
+      ex_message = str(ex)
+      if 'Connection refused' in ex_message or 'standby RM' in ex_message:
+        LOG.info('Resource Manager not available, trying another RM: %s.' % ex)
         rm_ha = get_next_ha_yarncluster()
         if rm_ha is not None:
           config, api.resource_manager_api = rm_ha
@@ -224,6 +225,9 @@ class YarnApi(JobBrowserApi):
       filters['finalStatus'] = state_filters[kwargs['state']]
 
     json = self.resource_manager_api.apps(**filters)
+    if type(json) == str and 'This is standby RM' in json:
+      raise Exception(json)
+
     if json['apps']:
       jobs = [Application(app) for app in json['apps']['app']]
     else:

+ 3 - 1
desktop/libs/hadoop/src/hadoop/cluster.py

@@ -174,11 +174,13 @@ def get_next_ha_yarncluster():
     if config.SUBMIT_TO.get():
       rm = ResourceManagerApi(config.RESOURCE_MANAGER_API_URL.get(), config.SECURITY_ENABLED.get())
       if has_ha:
-        try:          
+        try:
           cluster_info = rm.cluster()
           if cluster_info['clusterInfo']['haState'] == 'ACTIVE':
             MR_NAME_CACHE = name
             LOG.warn('Picking RM HA: %s' % name)
+            from hadoop.yarn import resource_manager_api
+            resource_manager_api._api_cache = None # Reset cache
             return (config, rm)
           else:
             LOG.info('RM %s is not RUNNING, skipping it: %s' % (name, cluster_info))