Bläddra i källkod

HUE-7494 [core] RM HA does not seem to get persisted

Romain Rigaux 8 år sedan
förälder
incheckning
e71f9b6

+ 156 - 2
apps/jobbrowser/src/jobbrowser/tests.py

@@ -25,7 +25,7 @@ import unittest
 from django.contrib.auth.models import User
 from django.core.urlresolvers import reverse
 from nose.plugins.skip import SkipTest
-from nose.tools import assert_true, assert_false, assert_equal
+from nose.tools import assert_true, assert_false, assert_equal, assert_raises
 
 from desktop.lib.django_test_util import make_logged_in_client
 from desktop.lib.test_utils import grant_access, add_to_group
@@ -38,6 +38,7 @@ from liboozie.oozie_api_tests import OozieServerProvider
 from oozie.models import Workflow
 
 from jobbrowser import models, views
+from jobbrowser.api import get_api
 from jobbrowser.conf import SHARE_JOBS
 from jobbrowser.models import can_view_job, can_modify_job, Job, LinkJobLogs
 
@@ -512,6 +513,159 @@ class TestMapReduce2NoHadoop:
     assert_true('Kill operation is forbidden.' in response.content, response.content)
 
 
+
+class TestResourceManagerHaNoHadoop:
+
+  def setUp(self):
+    # Beware: Monkey patching
+    if not hasattr(resource_manager_api, 'old_get_resource_manager_api'):
+      resource_manager_api.old_ResourceManagerApi = resource_manager_api.ResourceManagerApi
+    if not hasattr(mapreduce_api, 'old_get_mapreduce_api'):
+      mapreduce_api.old_get_mapreduce_api = mapreduce_api.get_mapreduce_api
+    if not hasattr(history_server_api, 'old_get_history_server_api'):
+      history_server_api.old_get_history_server_api = history_server_api.get_history_server_api
+
+    self.c = make_logged_in_client(is_superuser=False)
+    grant_access("test", "test", "jobbrowser")
+    self.user = User.objects.get(username='test')
+
+    resource_manager_api.ResourceManagerApi =  MockResourceManagerHaApi
+    mapreduce_api.get_mapreduce_api = lambda username: MockMapreduceHaApi(username)
+    history_server_api.get_history_server_api = lambda username: HistoryServerHaApi(username)
+
+    self.finish = []
+
+  def tearDown(self):
+    resource_manager_api.ResourceManagerApi = getattr(resource_manager_api, 'old_ResourceManagerApi')
+    mapreduce_api.get_mapreduce_api = getattr(mapreduce_api, 'old_get_mapreduce_api')
+    history_server_api.get_history_server_api = getattr(history_server_api, 'old_get_history_server_api')
+
+    for f in self.finish:
+      f()
+
+
+  def test_failover_no_ha(self):
+    self.finish = [
+        YARN_CLUSTERS.set_for_testing({'default': {}}),
+
+        YARN_CLUSTERS['default'].SUBMIT_TO.set_for_testing(True),
+        YARN_CLUSTERS['default'].RESOURCE_MANAGER_API_URL.set_for_testing('rm_host_active'),
+        YARN_CLUSTERS['default'].HISTORY_SERVER_API_URL.set_for_testing('jhs_host'),
+        YARN_CLUSTERS['default'].SECURITY_ENABLED.set_for_testing(False),
+        YARN_CLUSTERS['default'].SSL_CERT_CA_VERIFY.set_for_testing(False),
+    ]
+
+    resource_manager_api.API_CACHE = None
+    api = get_api(self.user, jt=None)
+
+    api.get_jobs(self.user, username=self.user.username, state='running', text='')
+    assert_false(api.resource_manager_api.from_failover)
+
+    api.get_jobs(self.user, username=self.user.username, state='running', text='')
+    assert_false(api.resource_manager_api.from_failover)
+
+    assert_raises(Exception, api.get_jobs, self.user, username=self.user.username, state='running', text='')
+
+
+  def test_failover_ha(self):
+    self.finish = [
+        YARN_CLUSTERS.set_for_testing({'ha1': {}, 'ha2': {}}),
+
+        YARN_CLUSTERS['ha1'].SUBMIT_TO.set_for_testing(True),
+        YARN_CLUSTERS['ha1'].RESOURCE_MANAGER_API_URL.set_for_testing('rm_host_active'),
+        YARN_CLUSTERS['ha1'].HISTORY_SERVER_API_URL.set_for_testing('jhs_host'),
+        YARN_CLUSTERS['ha1'].SECURITY_ENABLED.set_for_testing(False),
+        YARN_CLUSTERS['ha1'].SSL_CERT_CA_VERIFY.set_for_testing(False),
+        YARN_CLUSTERS['ha2'].SUBMIT_TO.set_for_testing(True),
+        YARN_CLUSTERS['ha2'].RESOURCE_MANAGER_API_URL.set_for_testing('rm_2_host'),
+        YARN_CLUSTERS['ha2'].HISTORY_SERVER_API_URL.set_for_testing('jhs_host'),
+        YARN_CLUSTERS['ha2'].SECURITY_ENABLED.set_for_testing(False),
+        YARN_CLUSTERS['ha2'].SSL_CERT_CA_VERIFY.set_for_testing(False),
+    ]
+
+
+    resource_manager_api.API_CACHE = None
+    api = get_api(self.user, jt=None)
+
+    api.get_jobs(self.user, username=self.user.username, state='running', text='')
+    assert_false(api.resource_manager_api.from_failover)
+
+    api.get_jobs(self.user, username=self.user.username, state='running', text='')
+    assert_false(api.resource_manager_api.from_failover)
+
+    # rm1 is set to to fail the 3rd time
+    YARN_CLUSTERS['ha1'].RESOURCE_MANAGER_API_URL.set_for_testing('rm_1_host')
+    YARN_CLUSTERS['ha2'].RESOURCE_MANAGER_API_URL.set_for_testing('rm_2_host_active') # Just tells mocked RM that it should say it is active
+    api.get_jobs(self.user, username=self.user.username, state='running', text='')
+    assert_true(api.resource_manager_api.from_failover)
+    api.resource_manager_api.from_failover = False
+
+    api.get_jobs(self.user, username=self.user.username, state='running', text='')
+    assert_false(api.resource_manager_api.from_failover)
+
+    # rm2 is set to to fail the 3rd time
+    YARN_CLUSTERS['ha1'].RESOURCE_MANAGER_API_URL.set_for_testing('rm_1_host_active')
+    YARN_CLUSTERS['ha2'].RESOURCE_MANAGER_API_URL.set_for_testing('rm_2_host')
+    api.get_jobs(self.user, username=self.user.username, state='running', text='')
+    assert_true(api.resource_manager_api.from_failover)
+    api.resource_manager_api.from_failover = False
+
+    api.get_jobs(self.user, username=self.user.username, state='running', text='')
+    assert_false(api.resource_manager_api.from_failover)
+
+    # if rm fails and no other active ones we fail
+    assert_raises(Exception, api.get_jobs, self.user, username=self.user.username, state='running', text='')
+
+
+class MockResourceManagerHaApi(object):
+  """
+  Mock the RM API.
+  Raise a failover exception after 2 calls. Is active if name contains 'active'.
+  """
+  def __init__(self, rm_url, security_enabled=False, ssl_cert_ca_verify=False):
+    self.rm_url = rm_url
+    self.from_failover = False
+    self.get_apps_count = 0
+
+  def setuser(self, user):
+    return user
+
+  @property
+  def user(self):
+    return 'test'
+
+  @property
+  def username(self):
+    return 'test'
+
+  @property
+  def url(self):
+    return self.rm_url
+
+  def apps(self, **kwargs):
+    if self.get_apps_count >= 2:
+      self.get_apps_count = 0
+      raise Exception('standby RM after 2 tries')
+
+    self.get_apps_count += 1
+    return {
+      'apps': {
+         'app': []
+      }
+    }
+
+  def cluster(self):
+    return {'clusterInfo': {'haState': 'ACTIVE' if 'active' in self.rm_url else 'STANDBY'}}
+
+
+class MockMapreduceHaApi(object):
+  def __init__(self, username): pass
+
+
+class HistoryServerHaApi(object):
+  def __init__(self, username): pass
+
+
 class MockResourceManagerApi:
   APPS = {
     'application_1356251510842_0054': {
@@ -797,7 +951,7 @@ class HistoryServerApi(MockMapreduce2Api):
               u'avgShuffleTime': 1421, u'queue': u'default', u'killedReduceAttempts': 0, u'failedMapAttempts': 0
           }
       }
-    else:      
+    else:
       return {
           u'job': {
               u'reducesCompleted': 1, u'avgMapTime': 1798, u'avgMergeTime': 1479, u'id': u'job_1356251510842_0009',

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

@@ -75,6 +75,7 @@ def rm_ha(funct):
         if rm_ha is not None:
           if rm_ha[1].url == api.resource_manager_api.url:
             raise ex
+          LOG.info('Retrying with Resource Manager: %s.' % rm_ha[1].url)
           config, api.resource_manager_api = rm_ha
           return funct(api, *args, **kwargs)
       raise ex
@@ -214,8 +215,6 @@ def get_next_ha_yarncluster(current_user=None):
   """
   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
 
@@ -233,15 +232,14 @@ def get_next_ha_yarncluster(current_user=None):
         try:
           cluster_info = rm.cluster()
           if cluster_info['clusterInfo']['haState'] == 'ACTIVE':
+            if name != MR_NAME_CACHE:
+              LOG.info('RM %s has failed back to %s server' % (MR_NAME_CACHE, name))
+              rm.from_failover = True
             MR_NAME_CACHE = name
             LOG.warn('Picking RM HA: %s' % name)
-            resource_manager_api.API_CACHE = None  # Reset cache
-            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:

+ 3 - 2
desktop/libs/hadoop/src/hadoop/yarn/resource_manager_api.py

@@ -39,8 +39,10 @@ _JSON_CONTENT_TYPE = 'application/json'
 API_CACHE = None
 API_CACHE_LOCK = threading.Lock()
 
+
 def get_resource_manager(username=None):
   global API_CACHE
+
   if API_CACHE is None:
     API_CACHE_LOCK.acquire()
     try:
@@ -56,8 +58,6 @@ def get_resource_manager(username=None):
 
   return API_CACHE
 
-class YarnFailoverOccurred(Exception):
-  pass
 
 class ResourceManagerApi(object):
 
@@ -67,6 +67,7 @@ class ResourceManagerApi(object):
     self._root = Resource(self._client)
     self._security_enabled = security_enabled
     self._thread_local = threading.local() # To store user info
+    self.from_failover = False
 
     if self._security_enabled:
       self._client.set_kerberos_auth()