|
@@ -18,6 +18,7 @@
|
|
|
import json
|
|
import json
|
|
|
import logging
|
|
import logging
|
|
|
import posixpath
|
|
import posixpath
|
|
|
|
|
+import Queue
|
|
|
import threading
|
|
import threading
|
|
|
|
|
|
|
|
from django.utils.translation import ugettext as _
|
|
from django.utils.translation import ugettext as _
|
|
@@ -36,31 +37,42 @@ LOG = logging.getLogger(__name__)
|
|
|
_API_VERSION = 'v1'
|
|
_API_VERSION = 'v1'
|
|
|
_JSON_CONTENT_TYPE = 'application/json'
|
|
_JSON_CONTENT_TYPE = 'application/json'
|
|
|
|
|
|
|
|
-API_CACHE = None
|
|
|
|
|
|
|
+API_CACHE_POOL = None
|
|
|
API_CACHE_LOCK = threading.Lock()
|
|
API_CACHE_LOCK = threading.Lock()
|
|
|
|
|
|
|
|
-
|
|
|
|
|
-def get_resource_manager(username):
|
|
|
|
|
- global API_CACHE
|
|
|
|
|
- if API_CACHE is None:
|
|
|
|
|
|
|
+def get_resource_manager_pool():
|
|
|
|
|
+ global API_CACHE_POOL
|
|
|
|
|
+ if API_CACHE_POOL is None:
|
|
|
API_CACHE_LOCK.acquire()
|
|
API_CACHE_LOCK.acquire()
|
|
|
try:
|
|
try:
|
|
|
- if API_CACHE is None:
|
|
|
|
|
|
|
+ if API_CACHE_POOL is None:
|
|
|
yarn_cluster = cluster.get_cluster_conf_for_job_submission()
|
|
yarn_cluster = cluster.get_cluster_conf_for_job_submission()
|
|
|
if yarn_cluster is None:
|
|
if yarn_cluster is None:
|
|
|
raise PopupException(_('No Resource Manager are available.'))
|
|
raise PopupException(_('No Resource Manager are available.'))
|
|
|
- API_CACHE = ResourceManagerApi(yarn_cluster.RESOURCE_MANAGER_API_URL.get(), yarn_cluster.SECURITY_ENABLED.get(), yarn_cluster.SSL_CERT_CA_VERIFY.get())
|
|
|
|
|
|
|
+ API_CACHE_POOL = ResourceManagerApiPool(yarn_cluster.RESOURCE_MANAGER_API_URL.get(), yarn_cluster.SECURITY_ENABLED.get(), yarn_cluster.SSL_CERT_CA_VERIFY.get())
|
|
|
finally:
|
|
finally:
|
|
|
API_CACHE_LOCK.release()
|
|
API_CACHE_LOCK.release()
|
|
|
|
|
|
|
|
- API_CACHE.setuser(username) # Set the correct user
|
|
|
|
|
-
|
|
|
|
|
- return API_CACHE
|
|
|
|
|
-
|
|
|
|
|
|
|
+ return API_CACHE_POOL
|
|
|
|
|
|
|
|
class YarnFailoverOccurred(Exception):
|
|
class YarnFailoverOccurred(Exception):
|
|
|
pass
|
|
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):
|
|
class ResourceManagerApi(object):
|
|
|
|
|
|
|
@@ -160,21 +172,6 @@ class ResourceManagerApi(object):
|
|
|
response = None
|
|
response = None
|
|
|
try:
|
|
try:
|
|
|
response = function(*args, **kwargs)
|
|
response = function(*args, **kwargs)
|
|
|
- except RestException, e:
|
|
|
|
|
- # YARN-2605: Yarn does not use proper HTTP redirects when the standby RM has
|
|
|
|
|
- # failed back to the master RM.
|
|
|
|
|
- if e.code == 307 and e.message.startswith('This is standby RM'):
|
|
|
|
|
- LOG.info('Received YARN failover redirect response, attempting to resolve redirect.')
|
|
|
|
|
- try:
|
|
|
|
|
- kwargs.update({'allow_redirects': True})
|
|
|
|
|
- response = function(*args, **kwargs)
|
|
|
|
|
- except Exception, e:
|
|
|
|
|
- if response:
|
|
|
|
|
- raise YarnFailoverOccurred(response)
|
|
|
|
|
- else:
|
|
|
|
|
- raise PopupException(_('Failed to resolve YARN RM: %s') % e)
|
|
|
|
|
- else:
|
|
|
|
|
- raise PopupException(_('YARN RM returned a failed response: %s') % e)
|
|
|
|
|
-
|
|
|
|
|
|
|
+ except Exception, e:
|
|
|
|
|
+ raise PopupException(_('YARN RM returned a failed response: %s') % e)
|
|
|
return response
|
|
return response
|
|
|
-
|
|
|