Browse Source

HUE-7258 [jb] Add config check for Spark history server URL

Ying Chen 7 years ago
parent
commit
56b665d

+ 38 - 1
desktop/libs/hadoop/src/hadoop/conf.py

@@ -118,6 +118,23 @@ MR_CLUSTERS = UnspecifiedConfigSection(
   )
 )
 
+
+def get_spark_history_server_from_cm():
+  from metadata.conf import MANAGER
+  from metadata.manager_client import ManagerApi
+
+  if MANAGER.API_URL.get():
+    return ManagerApi().get_spark_history_server_url()
+  return None
+
+def get_spark_history_server_url():
+  """
+    Try to get Spark history server URL from Cloudera Manager API, otherwise give default URL
+  """
+  url = get_spark_history_server_from_cm()
+  return url if url else 'http://localhost:18088'
+
+
 YARN_CLUSTERS = UnspecifiedConfigSection(
   "yarn_clusters",
   help="One entry for each Yarn cluster",
@@ -151,7 +168,7 @@ YARN_CLUSTERS = UnspecifiedConfigSection(
                   default='http://localhost:19888',
                   help="URL of the HistoryServer API"),
       SPARK_HISTORY_SERVER_URL=Config("spark_history_server_url",
-                  default='http://localhost:18088',
+                  dynamic_default=get_spark_history_server_url,
                   help="URL of the Spark History Server"),
       SSL_CERT_CA_VERIFY=Config("ssl_cert_ca_verify",
                   help="In secure mode (HTTPS), if SSL certificates from YARN Rest APIs have to be verified against certificate authority",
@@ -195,9 +212,29 @@ def config_validator(user):
   else:
     res.extend(test_yarn_configurations(user))
 
+  if get_spark_history_server_from_cm():
+    status = test_spark_configuration(user)
+    if status != 'OK':
+      res.append(("Spark_history_server", "Spark job can't retrieve logs of driver and executors without "
+                  "a running Spark history server"))
+
   return res
 
 
+def test_spark_configuration(user):
+  import hadoop.yarn.spark_history_server_api as spark_hs_api
+
+  status = None
+
+  try:
+    spark_hs_api.get_history_server_api().applications()
+    status = 'OK'
+  except:
+    LOG.exception('failed to get spark history server status')
+
+  return status
+
+
 def test_yarn_configurations(user):
   result = []
 

+ 71 - 0
desktop/libs/metadata/src/metadata/manager_client.py

@@ -16,9 +16,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import base64
 import json
 import logging
 import urllib
+import urllib2
 
 from django.core.cache import cache
 from django.utils.translation import ugettext as _
@@ -80,6 +82,75 @@ class ManagerApi(object):
       raise ManagerApiException(e)
 
 
+  def get_spark_history_server_url(self, cluster_name=None):
+    service_name = "SPARK_ON_YARN"
+    shs_role_type = "SPARK_YARN_HISTORY_SERVER"
+
+    try:
+      cluster = self._get_cluster(cluster_name)
+      services = self._root.get('clusters/%(cluster_name)s/services' % {
+        'cluster_name': cluster['name'],
+        'service_name': service_name
+      })['items']
+
+      service_display_names = [service['displayName'] for service in services if service['type'] == service_name]
+
+
+      if service_display_names:
+        spark_service_display_name = service_display_names[0]
+
+        servers = self._root.get('clusters/%(cluster_name)s/services/%(spark_service_display_name)s/roles' % {
+          'cluster_name': cluster['name'],
+          'spark_service_display_name': spark_service_display_name
+        })['items']
+
+        shs_server_names = [server['name'] for server in servers if server['type'] == shs_role_type]
+        shs_server_name = shs_server_names[0] if shs_server_names else None
+        shs_server_hostRef = [server['hostRef'] for server in servers if server['type'] == shs_role_type]
+        shs_server_hostId = shs_server_hostRef[0]['hostId'] if shs_server_hostRef else None
+
+        if shs_server_name and shs_server_hostId:
+          shs_server_configs = self._root.get('clusters/%(cluster_name)s/services/%(spark_service_display_name)s/roles/%(shs_server_name)s/config' % {
+            'cluster_name': cluster['name'],
+            'spark_service_display_name': spark_service_display_name,
+            'shs_server_name': shs_server_name
+          }, params={'view': 'full'})['items']
+
+          shs_ui_port = None
+          shs_ssl_port = None
+          shs_ssl_enabled = None
+          for config in shs_server_configs:
+            if 'relatedName' in config and 'default' in config:
+              if config['relatedName'] == 'spark.history.ui.port':
+                shs_ui_port = config['default']
+              if config['relatedName'] == 'spark.ssl.historyServer.port':
+                shs_ssl_port = config['default']
+              if config['relatedName'] == 'spark.ssl.historyServer.enabled':
+                shs_ssl_enabled = config['default']
+          shs_ui_host = self._root.get('hosts/%(hostId)s' % {'hostId': shs_server_hostId})
+          shs_ui_hostname = shs_ui_host['hostname'] if shs_ui_host else None
+
+          return self.assemble_shs_url(shs_ui_hostname, shs_ui_port, shs_ssl_port, shs_ssl_enabled)
+
+    except Exception, e:
+      LOG.warn("Check Spark history server via ManangerAPI: %s" % e)
+
+    return None
+
+  def assemble_shs_url(self, shs_ui_hostname, shs_ui_port=None, shs_ssl_port=None, shs_ssl_enabled=None):
+    if not shs_ui_hostname or not shs_ui_port or not shs_ssl_port or not shs_ssl_enabled:
+      LOG.warn("Spark conf not found!")
+      return None
+
+    protocol = 'https' if shs_ssl_enabled.lower() == 'true' else 'http'
+    shs_url = '%(protocol)s://%(hostname)s:%(port)s' % {
+      'protocol': protocol,
+      'hostname': shs_ui_hostname,
+      'port': shs_ssl_port if shs_ssl_enabled.lower() == 'true' else shs_ui_port,
+    }
+
+    return shs_url
+
   def tools_echo(self):
     try:
       params = (