|
@@ -22,6 +22,7 @@ import threading
|
|
|
|
|
|
|
|
from django.utils.translation import ugettext as _
|
|
from django.utils.translation import ugettext as _
|
|
|
|
|
|
|
|
|
|
+from desktop.conf import DEFAULT_USER
|
|
|
from desktop.lib.exceptions_renderable import PopupException
|
|
from desktop.lib.exceptions_renderable import PopupException
|
|
|
from desktop.lib.rest.http_client import HttpClient
|
|
from desktop.lib.rest.http_client import HttpClient
|
|
|
from desktop.lib.rest.resource import Resource
|
|
from desktop.lib.rest.resource import Resource
|
|
@@ -30,8 +31,8 @@ from hadoop import cluster
|
|
|
|
|
|
|
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
LOG = logging.getLogger(__name__)
|
|
|
-DEFAULT_USER = 'hue'
|
|
|
|
|
|
|
|
|
|
|
|
+DEFAULT_USER = DEFAULT_USER.get()
|
|
|
_API_VERSION = 'v1'
|
|
_API_VERSION = 'v1'
|
|
|
_JSON_CONTENT_TYPE = 'application/json'
|
|
_JSON_CONTENT_TYPE = 'application/json'
|
|
|
|
|
|
|
@@ -39,7 +40,7 @@ _api_cache = None
|
|
|
_api_cache_lock = threading.Lock()
|
|
_api_cache_lock = threading.Lock()
|
|
|
|
|
|
|
|
|
|
|
|
|
-def get_resource_manager():
|
|
|
|
|
|
|
+def get_resource_manager(user):
|
|
|
global _api_cache
|
|
global _api_cache
|
|
|
if _api_cache is None:
|
|
if _api_cache is None:
|
|
|
_api_cache_lock.acquire()
|
|
_api_cache_lock.acquire()
|
|
@@ -48,7 +49,7 @@ def get_resource_manager():
|
|
|
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 = ResourceManagerApi(user, 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()
|
|
|
return _api_cache
|
|
return _api_cache
|
|
@@ -60,8 +61,9 @@ class YarnFailoverOccurred(Exception):
|
|
|
|
|
|
|
|
class ResourceManagerApi(object):
|
|
class ResourceManagerApi(object):
|
|
|
|
|
|
|
|
- def __init__(self, oozie_url, security_enabled=False, ssl_cert_ca_verify=False):
|
|
|
|
|
- self._url = posixpath.join(oozie_url, 'ws', _API_VERSION)
|
|
|
|
|
|
|
+ def __init__(self, user, rm_url, security_enabled=False, ssl_cert_ca_verify=False):
|
|
|
|
|
+ self._user = user
|
|
|
|
|
+ self._url = posixpath.join(rm_url, 'ws', _API_VERSION)
|
|
|
self._client = HttpClient(self._url, logger=LOG)
|
|
self._client = HttpClient(self._url, logger=LOG)
|
|
|
self._root = Resource(self._client)
|
|
self._root = Resource(self._client)
|
|
|
self._security_enabled = security_enabled
|
|
self._security_enabled = security_enabled
|
|
@@ -71,6 +73,16 @@ class ResourceManagerApi(object):
|
|
|
|
|
|
|
|
self._client.set_verify(ssl_cert_ca_verify)
|
|
self._client.set_verify(ssl_cert_ca_verify)
|
|
|
|
|
|
|
|
|
|
+ def _get_params(self):
|
|
|
|
|
+ params = {}
|
|
|
|
|
+
|
|
|
|
|
+ if DEFAULT_USER != self._user.username: # We impersonate if needed
|
|
|
|
|
+ params['doAs'] = self._user.username
|
|
|
|
|
+ if not self.security_enabled:
|
|
|
|
|
+ params['user.name'] = DEFAULT_USER
|
|
|
|
|
+
|
|
|
|
|
+ return params
|
|
|
|
|
+
|
|
|
def __str__(self):
|
|
def __str__(self):
|
|
|
return "ResourceManagerApi at %s" % (self._url,)
|
|
return "ResourceManagerApi at %s" % (self._url,)
|
|
|
|
|
|
|
@@ -83,16 +95,22 @@ class ResourceManagerApi(object):
|
|
|
return self._security_enabled
|
|
return self._security_enabled
|
|
|
|
|
|
|
|
def cluster(self, **kwargs):
|
|
def cluster(self, **kwargs):
|
|
|
- return self._execute(self._root.get, 'cluster/info', params=kwargs, headers={'Accept': _JSON_CONTENT_TYPE})
|
|
|
|
|
|
|
+ params = self._get_params()
|
|
|
|
|
+ params.update(kwargs)
|
|
|
|
|
+ return self._execute(self._root.get, 'cluster/info', params=params, headers={'Accept': _JSON_CONTENT_TYPE})
|
|
|
|
|
|
|
|
def apps(self, **kwargs):
|
|
def apps(self, **kwargs):
|
|
|
- return self._execute(self._root.get, 'cluster/apps', params=kwargs, headers={'Accept': _JSON_CONTENT_TYPE})
|
|
|
|
|
|
|
+ params = self._get_params()
|
|
|
|
|
+ params.update(kwargs)
|
|
|
|
|
+ return self._execute(self._root.get, 'cluster/apps', params=params, headers={'Accept': _JSON_CONTENT_TYPE})
|
|
|
|
|
|
|
|
def app(self, app_id):
|
|
def app(self, app_id):
|
|
|
- return self._execute(self._root.get, 'cluster/apps/%(app_id)s' % {'app_id': app_id}, headers={'Accept': _JSON_CONTENT_TYPE})
|
|
|
|
|
|
|
+ params = self._get_params()
|
|
|
|
|
+ return self._execute(self._root.get, 'cluster/apps/%(app_id)s' % {'app_id': app_id}, params=params, headers={'Accept': _JSON_CONTENT_TYPE})
|
|
|
|
|
|
|
|
def kill(self, app_id):
|
|
def kill(self, app_id):
|
|
|
- return self._execute(self._root.put, 'cluster/apps/%(app_id)s/state' % {'app_id': app_id}, data=json.dumps({'state': 'KILLED'}), contenttype=_JSON_CONTENT_TYPE)
|
|
|
|
|
|
|
+ params = self._get_params()
|
|
|
|
|
+ return self._execute(self._root.put, 'cluster/apps/%(app_id)s/state' % {'app_id': app_id}, params=params, data=json.dumps({'state': 'KILLED'}), contenttype=_JSON_CONTENT_TYPE)
|
|
|
|
|
|
|
|
def _execute(self, function, *args, **kwargs):
|
|
def _execute(self, function, *args, **kwargs):
|
|
|
response = function(*args, **kwargs)
|
|
response = function(*args, **kwargs)
|