Эх сурвалжийг харах

[hadoop] Handle when the standby YARN RM fails back to the master RM

Erick Tryzelaar 10 жил өмнө
parent
commit
822afd4

+ 4 - 2
desktop/libs/hadoop/src/hadoop/cluster.py

@@ -175,6 +175,8 @@ def get_next_ha_yarncluster():
   """
   Return the next available YARN RM instance and cache its name.
   """
+  from hadoop.yarn import mapreduce_api
+  from hadoop.yarn import resource_manager_api
   from hadoop.yarn.resource_manager_api import ResourceManagerApi
   global MR_NAME_CACHE
 
@@ -190,13 +192,13 @@ def get_next_ha_yarncluster():
           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
-            from hadoop.yarn import mapreduce_api
             mapreduce_api._api_cache = None
             return (config, rm)
           else:
             LOG.info('RM %s is not RUNNING, skipping it: %s' % (name, cluster_info))
+        except resource_manager_api.YarnFailoverOccurred:
+          LOG.info('RM %s has failed back to another server' % (name,))
         except Exception, ex:
           LOG.exception('RM %s is not available, skipping it: %s' % (name, ex))
       else:

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

@@ -54,6 +54,10 @@ def get_resource_manager():
   return _api_cache
 
 
+class YarnFailoverOccurred(Exception):
+  pass
+
+
 class ResourceManagerApi(object):
 
   def __init__(self, oozie_url, security_enabled=False, ssl_cert_ca_verify=False):
@@ -80,12 +84,26 @@ class ResourceManagerApi(object):
 
   def cluster(self, **kwargs):
     return self._root.get('cluster', params=kwargs, headers={'Accept': _JSON_CONTENT_TYPE})
+    return self._execute(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})
+    return self._execute(self._root.get, 'cluster/apps', params=kwargs, headers={'Accept': _JSON_CONTENT_TYPE})
 
   def app(self, app_id):
     return self._root.get('cluster/apps/%(app_id)s' % {'app_id': app_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+    return self._execute(self._root.get, 'cluster/apps/%(app_id)s' % {'app_id': app_id}, headers={'Accept': _JSON_CONTENT_TYPE})
 
   def kill(self, app_id):
     return self._root.put('cluster/apps/%(app_id)s/state' % {'app_id': app_id}, data=json.dumps({'state': 'KILLED'}), contenttype=_JSON_CONTENT_TYPE)
+    return self._execute(self._root.put, 'cluster/apps/%(app_id)s/state' % {'app_id': app_id}, data=json.dumps({'state': 'KILLED'}), contenttype=_JSON_CONTENT_TYPE)
+
+  def _execute(self, function, *args, **kwargs):
+    response = function(*args, **kwargs)
+
+    # YARN-2605: Yarn does not use proper HTTP redirects when the standby RM has
+    # failed back to the master RM.
+    if isinstance(response, str) and response.startswith('This is standby RM. Redirecting to the current active RM'):
+      raise YarnFailoverOccurred(response)
+
+    return response