浏览代码

HUE-5115 [core] Revert custom job browser connection pooling

Prakash Ranade 9 年之前
父节点
当前提交
3ac4d68

+ 1 - 5
apps/jobbrowser/src/jobbrowser/api.py

@@ -174,15 +174,11 @@ class YarnApi(JobBrowserApi):
   """
   def __init__(self, user):
     self.user = user
-    self.resource_manager_api_pool = resource_manager_api.get_resource_manager_pool()
-    self.resource_manager_api = self.resource_manager_api_pool.get(user.username)
+    self.resource_manager_api = resource_manager_api.get_resource_manager(user.username)
     self.mapreduce_api = mapreduce_api.get_mapreduce_api(user.username)
     self.history_server_api = history_server_api.get_history_server_api(user.username)
     self.spark_history_server_api = spark_history_server_api.get_history_server_api()  # Spark HS does not support setuser
 
-  def __del__(self):
-    self.resource_manager_api_pool.put(self.resource_manager_api)
-
   def get_job_link(self, job_id):
     return self.get_job(job_id)
 

+ 5 - 33
apps/jobbrowser/src/jobbrowser/tests.py

@@ -390,8 +390,8 @@ class TestMapReduce2NoHadoop:
 
   def setUp(self):
     # Beware: Monkey patching
-    if not hasattr(resource_manager_api, 'old_get_resource_manager_pool'):
-      resource_manager_api.old_get_resource_manager_pool = resource_manager_api.get_resource_manager_pool
+    if not hasattr(resource_manager_api, 'old_get_resource_manager_api'):
+      resource_manager_api.old_get_resource_manager = resource_manager_api.get_resource_manager
     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'):
@@ -405,7 +405,7 @@ class TestMapReduce2NoHadoop:
     grant_access("test2", "test2", "jobbrowser")
     self.user2 = User.objects.get(username='test2')
 
-    resource_manager_api.get_resource_manager_pool = lambda: MockResourceManagerApiPool()
+    resource_manager_api.get_resource_manager = lambda username: MockResourceManagerApi(username)
     mapreduce_api.get_mapreduce_api = lambda username: MockMapreduceApi(username)
     history_server_api.get_history_server_api = lambda username: HistoryServerApi(username)
 
@@ -417,7 +417,7 @@ class TestMapReduce2NoHadoop:
 
 
   def tearDown(self):
-    resource_manager_api.get_resource_manager_pool = getattr(resource_manager_api, 'old_get_resource_manager_pool')
+    resource_manager_api.get_resource_manager = getattr(resource_manager_api, 'old_get_resource_manager')
     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')
 
@@ -502,23 +502,6 @@ class TestMapReduce2NoHadoop:
     finally:
       MockResourceManagerApi.APPS[job_id]['state'] = 'RUNNING'
 
-class MockResourceManagerApiPool:
-
-  def __init__(self):
-    pool_size = 10
-    self.rmobj_pool = Queue.LifoQueue()
-    for i in range(pool_size):
-      rm_instance = MockResourceManagerApi()
-      self.rmobj_pool.put(rm_instance)
-
-  def get(self, user):
-    rmobj = self.rmobj_pool.get()
-    rmobj.setuser(user)
-    return rmobj
-
-  def put(self, rmobj):
-    self.rmobj_pool.put(rmobj)
-
 class MockResourceManagerApi:
   APPS = {
     'application_1356251510842_0054': {
@@ -620,7 +603,7 @@ class MockResourceManagerApi:
     }
   }
 
-  def __init__(self, rm_url=None): pass
+  def __init__(self, user, rm_url=None): pass
 
   def apps(self, **kwargs):
     return {
@@ -643,17 +626,6 @@ class MockResourceManagerApi:
       u'app': MockResourceManagerApi.APPS[job_id]
     }
 
-  def setuser(self, user):
-    self.user = user
-    return user
-
-  @property
-  def user(self):
-    return self.username
-
-  @property
-  def username(self):
-    return self.user.username
 
 class MockMapreduce2Api(object):
   """

+ 1 - 3
apps/jobbrowser/src/jobbrowser/views.py

@@ -89,10 +89,8 @@ def get_job(request, job_id):
     job = get_api(request.user, request.jt).get_job(jobid=job_id)
   except ApplicationNotRunning, e:
     if e.job.get('state', '').lower() == 'accepted':
-      rm_pool = resource_manager_api.get_resource_manager_pool()
-      rm_api = rm_pool.get(request.user.username)
+      rm_api = resource_manager_api.get_resource_manager(request.user)
       job = Application(e.job, rm_api)
-      rm_pool.put(rm_api)
     else:
       raise e  # Job has not yet been accepted by RM
   except JobExpired, e:

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

@@ -235,6 +235,7 @@ def get_next_ha_yarncluster(current_user=None):
           if cluster_info['clusterInfo']['haState'] == 'ACTIVE':
             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:

+ 2 - 5
desktop/libs/hadoop/src/hadoop/yarn/mapreduce_api.py

@@ -24,7 +24,7 @@ from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.rest.http_client import HttpClient
 from desktop.lib.rest.resource import Resource
 from hadoop import cluster
-from hadoop.yarn.resource_manager_api import get_resource_manager_pool
+from hadoop.yarn.resource_manager_api import get_resource_manager
 
 
 LOG = logging.getLogger(__name__)
@@ -143,7 +143,4 @@ class MapreduceApi(object):
 
   def kill(self, job_id):
     app_id = job_id.replace('job', 'application')
-    pool = get_resource_manager_pool()
-    rmobj = pool.get(self.username)
-    rmobj.kill(app_id) # We need to call the RM
-    pool.put(rmobj)
+    get_resource_manager(self.username).kill(app_id) # We need to call the RM

+ 9 - 24
desktop/libs/hadoop/src/hadoop/yarn/resource_manager_api.py

@@ -18,7 +18,6 @@
 import json
 import logging
 import posixpath
-import Queue
 import threading
 
 from django.utils.translation import ugettext as _
@@ -37,43 +36,29 @@ LOG = logging.getLogger(__name__)
 _API_VERSION = 'v1'
 _JSON_CONTENT_TYPE = 'application/json'
 
-API_CACHE_POOL = None
+API_CACHE = None
 API_CACHE_LOCK = threading.Lock()
 
-def get_resource_manager_pool():
-  global API_CACHE_POOL
-  if API_CACHE_POOL is None:
+def get_resource_manager(username=None):
+  global API_CACHE
+  if API_CACHE is None:
     API_CACHE_LOCK.acquire()
     try:
-      if API_CACHE_POOL is None:
+      if API_CACHE is None:
         yarn_cluster = cluster.get_cluster_conf_for_job_submission()
         if yarn_cluster is None:
           raise PopupException(_('No Resource Manager are available.'))
-        API_CACHE_POOL = ResourceManagerApiPool(yarn_cluster.RESOURCE_MANAGER_API_URL.get(), yarn_cluster.SECURITY_ENABLED.get(), yarn_cluster.SSL_CERT_CA_VERIFY.get())
+        API_CACHE = ResourceManagerApi(yarn_cluster.RESOURCE_MANAGER_API_URL.get(), yarn_cluster.SECURITY_ENABLED.get(), yarn_cluster.SSL_CERT_CA_VERIFY.get())
     finally:
       API_CACHE_LOCK.release()
 
-  return API_CACHE_POOL
+  API_CACHE.setuser(username) # Set the correct user
+
+  return API_CACHE
 
 class YarnFailoverOccurred(Exception):
   pass
 
-class ResourceManagerApiPool(object):
-  def __init__(self, api_url, security_enabled, ssl_cert):
-    pool_size = 10
-    self.rmobj_pool = Queue.LifoQueue()
-    for i in range(pool_size):
-      rm_instance = ResourceManagerApi(api_url, security_enabled, ssl_cert)
-      self.rmobj_pool.put(rm_instance)
-
-  def get(self, username):
-    rmobj = self.rmobj_pool.get()
-    rmobj.setuser(username)
-    return rmobj
-
-  def put(self, rmobj):
-    self.rmobj_pool.put(rmobj)
-
 class ResourceManagerApi(object):
 
   def __init__(self, rm_url, security_enabled=False, ssl_cert_ca_verify=False):