Explorar o código

[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 %!s(int64=11) %!d(string=hai) anos
pai
achega
0109c9d

+ 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))