Explorar o código

HUE-4258 [jb] Close and pool Spark History Server connections

Jenny Kim %!s(int64=9) %!d(string=hai) anos
pai
achega
53cc6f0ddd

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

@@ -177,7 +177,7 @@ class YarnApi(JobBrowserApi):
     self.resource_manager_api = resource_manager_api.get_resource_manager(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.mapreduce_api = mapreduce_api.get_mapreduce_api(user.username)
     self.history_server_api = history_server_api.get_history_server_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()
+    self.spark_history_server_api = spark_history_server_api.get_history_server_api()  # Spark HS does not support setuser
 
 
   def get_job_link(self, job_id):
   def get_job_link(self, job_id):
     return self.get_job(job_id)
     return self.get_job(job_id)

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

@@ -123,14 +123,18 @@ class SparkJob(Application):
     return self.trackingUrl.strip('/').split('/')[-1]
     return self.trackingUrl.strip('/').split('/')[-1]
 
 
   def _resolve_tracking_url(self):
   def _resolve_tracking_url(self):
+    resp = None
     try:
     try:
-      resp = urllib2.urlopen(self.trackingUrl)
+      resp = urllib2.urlopen(self.trackingUrl, timeout=5.0)
       actual_url = resp.url
       actual_url = resp.url
       if actual_url.strip('/').split('/')[-1] == 'jobs':
       if actual_url.strip('/').split('/')[-1] == 'jobs':
         actual_url = actual_url.strip('/').replace('jobs', '')
         actual_url = actual_url.strip('/').replace('jobs', '')
       self.trackingUrl = actual_url
       self.trackingUrl = actual_url
     except Exception, e:
     except Exception, e:
       LOG.warn("Failed to resolve Spark Job's actual tracking URL: %s" % e)
       LOG.warn("Failed to resolve Spark Job's actual tracking URL: %s" % e)
+    finally:
+      if resp is not None:
+        resp.close()
 
 
   def _get_metrics(self):
   def _get_metrics(self):
     self.metrics = {}
     self.metrics = {}

+ 10 - 10
apps/spark/src/spark/job_server_api.py

@@ -34,21 +34,21 @@ _JSON_CONTENT_TYPE = 'application/json'
 _BINARY_CONTENT_TYPE = 'application/octet-stream'
 _BINARY_CONTENT_TYPE = 'application/octet-stream'
 _TEXT_CONTENT_TYPE = 'text/plain'
 _TEXT_CONTENT_TYPE = 'text/plain'
 
 
-_api_cache = None
-_api_cache_lock = threading.Lock()
+API_CACHE = None
+API_CACHE_LOCK = threading.Lock()
 
 
 
 
 def get_api(user):
 def get_api(user):
-  global _api_cache
-  if _api_cache is None:
-    _api_cache_lock.acquire()
+  global API_CACHE
+  if API_CACHE is None:
+    API_CACHE_LOCK.acquire()
     try:
     try:
-      if _api_cache is None:
-        _api_cache = JobServerApi(get_livy_server_url())
+      if API_CACHE is None:
+        API_CACHE = JobServerApi(get_livy_server_url())
     finally:
     finally:
-      _api_cache_lock.release()
-  _api_cache.setuser(user)
-  return _api_cache
+      API_CACHE_LOCK.release()
+  API_CACHE.setuser(user)
+  return API_CACHE
 
 
 
 
 class JobServerApi(object):
 class JobServerApi(object):

+ 0 - 3
desktop/libs/hadoop/src/hadoop/yarn/history_server_api.py

@@ -31,9 +31,6 @@ 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_lock = threading.Lock()
-
 API_CACHE = None
 API_CACHE = None
 API_CACHE_LOCK = threading.Lock()
 API_CACHE_LOCK = threading.Lock()
 
 

+ 15 - 10
desktop/libs/hadoop/src/hadoop/yarn/spark_history_server_api.py

@@ -19,32 +19,37 @@ import logging
 import posixpath
 import posixpath
 import threading
 import threading
 
 
+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
 from hadoop import cluster
 from hadoop import cluster
 
 
 
 
 LOG = logging.getLogger(__name__)
 LOG = logging.getLogger(__name__)
-DEFAULT_USER = 'hue'
 
 
 _API_VERSION = 'v1'
 _API_VERSION = 'v1'
 _JSON_CONTENT_TYPE = 'application/json'
 _JSON_CONTENT_TYPE = 'application/json'
 
 
-_api_cache = None
-_api_cache_lock = threading.Lock()
+API_CACHE = None
+API_CACHE_LOCK = threading.Lock()
 
 
 
 
 def get_history_server_api():
 def get_history_server_api():
-  global _api_cache
-  if _api_cache is None:
-    _api_cache_lock.acquire()
+  # TODO: Spark History Server does not yet support setuser, implement when it does
+  global API_CACHE
+
+  if API_CACHE is None:
+    API_CACHE_LOCK.acquire()
     try:
     try:
-      if _api_cache is None:
+      if API_CACHE is None:
         yarn_cluster = cluster.get_cluster_conf_for_job_submission()
         yarn_cluster = cluster.get_cluster_conf_for_job_submission()
-        _api_cache = SparkHistoryServerApi(yarn_cluster.SPARK_HISTORY_SERVER_URL.get(), yarn_cluster.SECURITY_ENABLED.get(), yarn_cluster.SSL_CERT_CA_VERIFY.get())
+        if yarn_cluster is None:
+          raise PopupException(_('No Spark History Server is available.'))
+        API_CACHE = SparkHistoryServerApi(yarn_cluster.SPARK_HISTORY_SERVER_URL.get(), yarn_cluster.SECURITY_ENABLED.get(), yarn_cluster.SSL_CERT_CA_VERIFY.get())
     finally:
     finally:
-      _api_cache_lock.release()
-  return _api_cache
+      API_CACHE_LOCK.release()
+
+  return API_CACHE
 
 
 
 
 class SparkHistoryServerApi(object):
 class SparkHistoryServerApi(object):