Ver código fonte

HUE-1905 [core] Add security handling for yarn and mr2

Reuse security_enabled in yarn config section for both yarn and MR2 security.
Add kerberos handler for all log calls and yarn/mr2 api clients.
If a particular service is not kerberized, it should still work
since SPNEGO seems to work with insecure clusters.
Abraham Elmahrek 12 anos atrás
pai
commit
8e441d8

+ 9 - 2
apps/jobbrowser/src/jobbrowser/views.py

@@ -20,6 +20,7 @@ import re
 import time
 import logging
 import string
+import urlparse
 from urllib import quote_plus
 from lxml import html
 
@@ -30,11 +31,13 @@ from django.core.urlresolvers import reverse
 
 from desktop.log.access import access_warn, access_log_level
 from desktop.lib.rest.http_client import RestException
+from desktop.lib.rest.resource import Resource
 from desktop.lib.django_util import render_json, render, copy_query_dict, encode_json_for_js
 from desktop.lib.exceptions import MessageException
 from desktop.lib.exceptions_renderable import PopupException
 from desktop.views import register_status_bar_view
 from hadoop.api.jobtracker.ttypes import ThriftJobPriority, TaskTrackerNotFoundException, ThriftJobState
+from hadoop.yarn.clients import get_log_client
 
 from jobbrowser import conf
 from jobbrowser.api import get_api, ApplicationNotRunning
@@ -233,11 +236,15 @@ def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=0
     raise KeyError(_("Cannot find job attempt '%(id)s'.") % {'id': job.jobId}, e)
 
   link = '/%s/' % name
+  params = {}
   if offset and int(offset) >= 0:
-    link += '?start=%s' % offset
+    params['start'] = offset
+
+  root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2])
 
   try:
-    log = html.parse(log_link + link).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
+    response = root.get(link, params=params)
+    log = html.fromstring(response).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
   except Exception, e:
     log = _('Failed to retrieve log: %s') % e
 

+ 11 - 2
apps/jobbrowser/src/jobbrowser/yarn_models.py

@@ -18,11 +18,17 @@
 import logging
 import re
 import time
+import urlparse
 
 from lxml import html
 
+from django.utils.translation import ugettext as _
+
+from desktop.lib.rest.resource import Resource
 from desktop.lib.view_util import format_duration_in_millis
 
+from hadoop.yarn.clients import get_log_client
+
 from jobbrowser.models import format_unixtime_ms
 
 
@@ -211,12 +217,15 @@ class Attempt:
 
     for name in ('stdout', 'stderr', 'syslog'):
       link = '/%s/' % name
+      params = {}
       if int(offset) >= 0:
-        link += '?start=%s' % offset
+        params['start'] = offset
 
       try:
         log_link = re.sub('job_[^/]+', self.id, log_link)
-        log = html.parse(log_link + link).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
+        root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2])
+        response = root.get(link, params=params)
+        log = html.fromstring(response).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
       except Exception, e:
         log = _('Failed to retrieve log: %s') % e
 

+ 68 - 0
desktop/libs/hadoop/src/hadoop/yarn/clients.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import threading
+import time
+import urlparse
+import heapq
+
+from desktop.lib.rest.http_client import HttpClient
+
+from hadoop import cluster
+
+
+LOG = logging.getLogger(__name__)
+
+MAX_HEAP_SIZE = 20
+
+_log_client_heap = []
+_log_client_lock = threading.Lock()
+
+
+def get_log_client(log_link):
+  global _log_client_queue
+  global MAX_HEAP_SIZE
+  _log_client_lock.acquire()
+  try:
+    components = urlparse.urlsplit(log_link)
+    base_url = '%(scheme)s://%(netloc)s' % {
+      'scheme': components[0],
+      'netloc': components[1]
+    }
+
+    # Takes on form (epoch time, client object)
+    # Least Recently Used algorithm.
+    client_tuple = next((tup for tup in _log_client_heap if tup[1].base_url == base_url), None)
+    if client_tuple is None:
+      client = HttpClient(base_url, LOG)
+      yarn_cluster = cluster.get_cluster_conf_for_job_submission()
+      if yarn_cluster.SECURITY_ENABLED.get():
+        client.set_kerberos_auth()
+    else:
+      _log_client_heap.remove(client_tuple)
+      client = client_tuple[1]
+
+    new_client_tuple = (time.time(), client)
+    if len(_log_client_heap) >= MAX_HEAP_SIZE:
+      heapq.heapreplace(_log_client_heap, new_client_tuple)
+    else:
+      heapq.heappush(_log_client_heap, new_client_tuple)
+
+    return client
+  finally:
+    _log_client_lock.release()

+ 6 - 2
desktop/libs/hadoop/src/hadoop/yarn/history_server_api.py

@@ -41,7 +41,7 @@ def get_history_server_api():
     try:
       if _api_cache is None:
         yarn_cluster = cluster.get_cluster_conf_for_job_submission()
-        _api_cache = HistoryServerApi(yarn_cluster.HISTORY_SERVER_API_URL.get())
+        _api_cache = HistoryServerApi(yarn_cluster.HISTORY_SERVER_API_URL.get(), yarn_cluster.SECURITY_ENABLED.get())
     finally:
       _api_cache_lock.release()
   return _api_cache
@@ -49,10 +49,14 @@ def get_history_server_api():
 
 class HistoryServerApi(object):
 
-  def __init__(self, oozie_url):
+  def __init__(self, oozie_url, security_enabled=False):
     self._url = posixpath.join(oozie_url, 'ws/%s/history' % _API_VERSION)
     self._client = HttpClient(self._url, logger=LOG)
     self._root = Resource(self._client)
+    self._security_enabled = security_enabled
+
+    if self._security_enabled:
+      self._client.set_kerberos_auth()
 
   def __str__(self):
     return "HistoryServerApi at %s" % (self._url,)

+ 6 - 3
desktop/libs/hadoop/src/hadoop/yarn/mapreduce_api.py

@@ -41,7 +41,7 @@ def get_mapreduce_api():
     try:
       if _api_cache is None:
         yarn_cluster = cluster.get_cluster_conf_for_job_submission()
-        _api_cache = MapreduceApi(yarn_cluster.PROXY_API_URL.get())
+        _api_cache = MapreduceApi(yarn_cluster.PROXY_API_URL.get(), yarn_cluster.SECURITY_ENABLED.get())
     finally:
       _api_cache_lock.release()
   return _api_cache
@@ -49,11 +49,14 @@ def get_mapreduce_api():
 
 class MapreduceApi(object):
 
-  def __init__(self, oozie_url):
+  def __init__(self, oozie_url, security_enabled=False):
     self._url = posixpath.join(oozie_url, 'proxy')
     self._client = HttpClient(self._url, logger=LOG)
     self._root = Resource(self._client)
-    self._security_enabled = False
+    self._security_enabled = security_enabled
+
+    if self._security_enabled:
+      self._client.set_kerberos_auth()
 
   def __str__(self):
     return "MapreduceApi at %s" % (self._url,)

+ 8 - 3
desktop/libs/hadoop/src/hadoop/yarn/node_manager_api.py

@@ -21,6 +21,8 @@ import posixpath
 from desktop.lib.rest.http_client import HttpClient
 from desktop.lib.rest.resource import Resource
 
+from hadoop import cluster
+
 
 LOG = logging.getLogger(__name__)
 DEFAULT_USER = 'hue'
@@ -31,15 +33,18 @@ _JSON_CONTENT_TYPE = 'application/json'
 
 
 def get_resource_manager_api(api_url):
-  return ResourceManagerApi(api_url)
+  return ResourceManagerApi(api_url, cluster.get_cluster_conf_for_job_submission().SECURITY_ENABLED.get())
 
 
 class ResourceManagerApi(object):
-  def __init__(self, oozie_url):
+  def __init__(self, oozie_url, security_enabled=False):
     self._url = posixpath.join(oozie_url, 'ws', _API_VERSION)
     self._client = HttpClient(self._url, logger=LOG)
     self._root = Resource(self._client)
-    self._security_enabled = False
+    self._security_enabled = security_enabled
+
+    if self._security_enabled:
+      self._client.set_kerberos_auth()
 
   def __str__(self):
     return "NodeManagerApi at %s" % (self._url,)

+ 6 - 3
desktop/libs/hadoop/src/hadoop/yarn/resource_manager_api.py

@@ -41,18 +41,21 @@ def get_resource_manager():
     try:
       if _api_cache is None:
         yarn_cluster = cluster.get_cluster_conf_for_job_submission()
-        _api_cache = ResourceManagerApi(yarn_cluster.RESOURCE_MANAGER_API_URL.get())
+        _api_cache = ResourceManagerApi(yarn_cluster.RESOURCE_MANAGER_API_URL.get(), yarn_cluster.SECURITY_ENABLED.get())
     finally:
       _api_cache_lock.release()
   return _api_cache
 
 
 class ResourceManagerApi(object):
-  def __init__(self, oozie_url):
+  def __init__(self, oozie_url, security_enabled=False):
     self._url = posixpath.join(oozie_url, 'ws', _API_VERSION)
     self._client = HttpClient(self._url, logger=LOG)
     self._root = Resource(self._client)
-    self._security_enabled = False
+    self._security_enabled = security_enabled
+
+    if self._security_enabled:
+      self._client.set_kerberos_auth()
 
   def __str__(self):
     return "ResourceManagerApi at %s" % (self._url,)

+ 27 - 0
desktop/libs/hadoop/src/hadoop/yarn/tests.py

@@ -17,7 +17,10 @@
 
 import logging
 
+from nose.tools import assert_true, assert_equal, assert_not_equal
+
 from hadoop.yarn.resource_manager_api import get_resource_manager
+from hadoop.yarn import clients
 
 
 LOG = logging.getLogger(__name__)
@@ -38,3 +41,27 @@ def test_yarn_configurations():
     result.append(('Resource Manager', msg))
 
   return result
+
+
+def test_get_log_client():
+  old_max_heap_size = clients.MAX_HEAP_SIZE
+  clients.MAX_HEAP_SIZE = 2
+  try:
+    log_link1 = "http://test1:8041/container/nonsense"
+    log_link2 = "http://test2:8041/container/nonsense"
+    log_link3 = "http://test3:8041/container/nonsense"
+
+    c1 = clients.get_log_client(log_link1)
+    c2 = clients.get_log_client(log_link2)
+
+    assert_not_equal(c1, c2)
+    assert_equal(c1, clients.get_log_client(log_link1))
+
+    clients.get_log_client(log_link3)
+
+    assert_equal(2, len(clients._log_client_heap))
+    base_urls = [tup[1].base_url for tup in clients._log_client_heap]
+    assert_true('http://test1:8041' in base_urls)
+    assert_true('http://test3:8041' in base_urls)
+  finally:
+    clients.MAX_HEAP_SIZE = old_max_heap_size