Browse Source

[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 năm trước cách đây
mục cha
commit
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))