Bläddra i källkod

HUE-6160 [sentry] Support Sentry HA with failover

Attempts to use new sentry.service.client.server.rpc_address property to obtain list of hosts, but will fallback to explicitly configured hostname and port if sentry-site is not configured.
Jenny Kim 8 år sedan
förälder
incheckning
edb28fc

+ 12 - 63
desktop/libs/libsentry/src/libsentry/api.py

@@ -16,28 +16,20 @@
 # limitations under the License.
 
 import logging
-import json
-import random
-import threading
 import time
 
 from django.utils.translation import ugettext as _
 
+from desktop.lib.exceptions import StructuredThriftTransportException
 from desktop.lib.exceptions_renderable import PopupException
-from libzookeeper.models import ZookeeperClient
 
 from libsentry.client import SentryClient
-from libsentry.conf import HOSTNAME, PORT
-from libsentry.sentry_site import get_sentry_server_ha_enabled, get_sentry_server_ha_zookeeper_quorum, get_sentry_server_ha_zookeeper_namespace
+from libsentry.sentry_site import get_sentry_client, is_ha_enabled
 
 
 LOG = logging.getLogger(__name__)
 
 
-_api_cache = None
-_api_cache_lock = threading.Lock()
-
-
 def ha_error_handler(func):
   def decorator(*args, **kwargs):
     retries = 15
@@ -45,26 +37,25 @@ def ha_error_handler(func):
     while retries > 0:
       try:
         return func(*args, **kwargs)
-      except SentryException, e:
-        raise e
-      except Exception, e:
+      except StructuredThriftTransportException, e:
         retries -= 1
-        if not get_sentry_server_ha_enabled() or retries == 0:
-          raise e
+        if not is_ha_enabled() or retries == 0:
+          raise PopupException(_('Failed to retry connecting to an available Sentry server.'), detail=e)
         else:
-          # Right now retries on any error and pull a fresh list of servers from ZooKeeper
-          LOG.info('Retrying fetching an available client in ZooKeeper.')
-          global _api_cache
-          _api_cache = None
+          LOG.info('Could not connect to Sentry server %s, attempting to fetch next available client.' % args[0].client.host)
           time.sleep(1)
-          args[0].client = _get_client(args[0].client.username)
+          args[0].client = get_sentry_client(args[0].client.username, SentryClient)
           LOG.info('Picked %s' % args[0].client)
+      except SentryException, e:
+        raise e
+      except Exception, e:
+        raise PopupException(_('Encountered unexpected error in SentryApi.'), detail=e)
 
   return decorator
 
 
 def get_api(user):
-  client = _get_client(user.username)
+  client = get_sentry_client(user.username, SentryClient)
 
   return SentryApi(client)
 
@@ -228,45 +219,3 @@ class SentryException(Exception):
 
   def __str__(self):
     return self.message
-
-
-def _get_client(username):
-  if get_sentry_server_ha_enabled():
-    servers = _get_server_properties()
-    if servers:
-      server = random.choice(servers)
-    else:
-      raise PopupException(_('No Sentry servers are available.'))
-  else:
-    server = {
-        'hostname': HOSTNAME.get(),
-        'port': PORT.get()
-    }
-
-  return SentryClient(server['hostname'], server['port'], username)
-
-
-def _get_server_properties():
-  global _api_cache
-
-  if not _api_cache: # If we need to refresh the list or if previously no servers were up 
-    _api_cache_lock.acquire()
-
-    try:
-      if not _api_cache:
-
-        servers = []
-        with ZookeeperClient(hosts=get_sentry_server_ha_zookeeper_quorum()) as client:
-          sentry_servers = client.get_children_data(namespace=get_sentry_server_ha_zookeeper_namespace())
-
-        for data in sentry_servers:
-          server = json.loads(data.decode("utf-8"))
-          servers.append({'hostname': server['address'], 'port': server['sslPort'] if server['sslPort'] else server['port']})
-
-        _api_cache = servers
-    except Exception, e:
-      raise PopupException(_('Error in retrieving Sentry server properties from Zookeeper.'), detail=e)
-    finally:
-      _api_cache_lock.release()
-
-  return _api_cache

+ 12 - 63
desktop/libs/libsentry/src/libsentry/api2.py

@@ -16,28 +16,20 @@
 # limitations under the License.
 
 import logging
-import json
-import random
-import threading
 import time
 
 from django.utils.translation import ugettext as _
 
+from desktop.lib.exceptions import StructuredThriftTransportException
 from desktop.lib.exceptions_renderable import PopupException
-from libzookeeper.models import ZookeeperClient
 
 from libsentry.client2 import SentryClient
-from libsentry.conf import HOSTNAME, PORT
-from libsentry.sentry_site import get_sentry_server_ha_enabled, get_sentry_server_ha_zookeeper_quorum, get_sentry_server_ha_zookeeper_namespace
+from libsentry.sentry_site import get_sentry_client, is_ha_enabled
 
 
 LOG = logging.getLogger(__name__)
 
 
-_api_cache = None
-_api_cache_lock = threading.Lock()
-
-
 def ha_error_handler(func):
   def decorator(*args, **kwargs):
     retries = 15
@@ -45,20 +37,19 @@ def ha_error_handler(func):
     while retries > 0:
       try:
         return func(*args, **kwargs)
-      except SentryException, e:
-        raise e
-      except Exception, e:
+      except StructuredThriftTransportException, e:
         retries -= 1
-        if not get_sentry_server_ha_enabled() or retries == 0:
-          raise e
+        if not is_ha_enabled() or retries == 0:
+          raise PopupException(_('Failed to retry connecting to an available Sentry server.'), detail=e)
         else:
-          # Right now retries on any error and pull a fresh list of servers from ZooKeeper
-          LOG.info('Retrying fetching an available client in ZooKeeper.')
-          global _api_cache
-          _api_cache = None
+          LOG.info('Could not connect to Sentry server %s, attempting to fetch next available client.' % args[0].client.host)
           time.sleep(1)
-          args[0].client = _get_client(args[0].client.username, args[0].client.component)
+          args[0].client = get_sentry_client(args[0].client.username, SentryClient)
           LOG.info('Picked %s' % args[0].client)
+      except SentryException, e:
+        raise e
+      except Exception, e:
+        raise PopupException(_('Encountered unexpected error in SentryApi.'), detail=e)
 
   return decorator
 
@@ -67,7 +58,7 @@ def get_api(user, component):
   if component == 'solr':
     component = component.upper()
 
-  client = _get_client(user.username, component)
+  client = get_sentry_client(user.username, SentryClient, component)
 
   return SentryApi(client)
 
@@ -227,45 +218,3 @@ class SentryException(Exception):
 
   def __str__(self):
     return self.message
-
-
-def _get_client(username, component):
-  if get_sentry_server_ha_enabled():
-    servers = _get_server_properties()
-    if servers:
-      server = random.choice(servers)
-    else:
-      raise PopupException(_('No Sentry servers are available.'))
-  else:
-    server = {
-        'hostname': HOSTNAME.get(),
-        'port': PORT.get()
-    }
-
-  return SentryClient(server['hostname'], server['port'], username, component)
-
-
-def _get_server_properties():
-  global _api_cache
-
-  if not _api_cache: # If we need to refresh the list or if previously no servers were up 
-    _api_cache_lock.acquire()
-
-    try:
-      if not _api_cache:
-
-        servers = []
-        with ZookeeperClient(hosts=get_sentry_server_ha_zookeeper_quorum()) as client:
-          sentry_servers = client.get_children_data(namespace=get_sentry_server_ha_zookeeper_namespace())
-
-        for data in sentry_servers:
-          server = json.loads(data.decode("utf-8"))
-          servers.append({'hostname': server['address'], 'port': server['sslPort'] if server['sslPort'] else server['port']})
-
-        _api_cache = servers
-    except Exception, e:
-      raise PopupException(_('Error in retrieving Sentry server properties from Zookeeper.'), detail=e)
-    finally:
-      _api_cache_lock.release()
-
-  return _api_cache

+ 61 - 14
desktop/libs/libsentry/src/libsentry/sentry_site.py

@@ -18,11 +18,16 @@
 import errno
 import logging
 import os.path
+import random
+
+from django.utils.translation import ugettext as _
 
 from hadoop import confparse
+
 from desktop.lib import security_util
+from desktop.lib.exceptions_renderable import PopupException
 
-from libsentry.conf import SENTRY_CONF_DIR, HOSTNAME
+from libsentry.conf import SENTRY_CONF_DIR, HOSTNAME, PORT
 
 
 LOG = logging.getLogger(__name__)
@@ -36,10 +41,8 @@ _CONF_SENTRY_SERVER_PRINCIPAL = 'sentry.service.server.principal'
 _CONF_SENTRY_SERVER_SECURITY_MODE = 'sentry.service.security.mode'
 _CONF_SENTRY_SERVER_ADMIN_GROUP = 'sentry.service.admin.group'
 
-_CONF_SENTRY_SERVER_HA_ENABLED = 'sentry.ha.enabled'
-_CONF_SENTRY_SERVER_HA_HAS_SECURITY = 'sentry.ha.zookeeper.security'
-_CONF_SENTRY_SERVER_HA_ZOOKEEPER_ADDRESSES = 'sentry.ha.zookeeper.quorum'
-_CONF_SENTRY_SERVER_HA_ZOOKEEPER_NAMESPACE = 'sentry.ha.zookeeper.namespace'
+_CONF_SENTRY_SERVER_RPC_ADDRESSES = 'sentry.service.client.server.rpc-address'
+_CONF_SENTRY_SERVER_RPC_PORT = 'sentry.service.client.server.rpc-port'
 
 
 def reset():
@@ -53,10 +56,10 @@ def get_conf(name='sentry'):
   return _SITE_DICT[name]
 
 
-
 def get_hive_sentry_provider():
   return get_conf(name='hive').get(_CONF_HIVE_PROVIDER, 'server1')
 
+
 def get_solr_sentry_provider():
   return 'service1'
 
@@ -70,24 +73,67 @@ def get_sentry_server_principal():
   else:
     return None
 
+
 def get_sentry_server_authentication():
   return get_conf().get(_CONF_SENTRY_SERVER_SECURITY_MODE, 'NOSASL').upper()
 
+
 def get_sentry_server_admin_groups():
   return get_conf().get(_CONF_SENTRY_SERVER_ADMIN_GROUP, '').split(',')
 
 
-def get_sentry_server_ha_enabled():
-  return get_conf().get(_CONF_SENTRY_SERVER_HA_ENABLED, 'FALSE').upper() == 'TRUE'
+def get_sentry_server_rpc_addresses():
+  hosts = None
+  servers = get_conf().get(_CONF_SENTRY_SERVER_RPC_ADDRESSES)
+  if servers:
+    hosts = servers.split(',')
+  return hosts
+
+
+def get_sentry_server_rpc_port():
+  return int(get_conf().get(_CONF_SENTRY_SERVER_RPC_PORT, '8038'))
 
-def get_sentry_server_ha_has_security():
-  return get_conf().get(_CONF_SENTRY_SERVER_HA_HAS_SECURITY, 'FALSE').upper() == 'TRUE'
 
-def get_sentry_server_ha_zookeeper_quorum():
-  return get_conf().get(_CONF_SENTRY_SERVER_HA_ZOOKEEPER_ADDRESSES)
+def is_ha_enabled():
+  return get_sentry_server_rpc_addresses() is not None
 
-def get_sentry_server_ha_zookeeper_namespace():
-  return get_conf().get(_CONF_SENTRY_SERVER_HA_ZOOKEEPER_NAMESPACE, 'sentry')
+
+def get_sentry_client(username, client_class, component=None):
+  server = None
+  if is_ha_enabled():
+    servers = _get_server_properties()
+    if servers:
+      server = random.choice(servers)
+
+  if server is None:
+    if HOSTNAME.get() and PORT.get():
+      LOG.info('No Sentry servers configured in %s, falling back to libsentry configured host: %s:%s' %
+               (_CONF_SENTRY_SERVER_RPC_ADDRESSES, HOSTNAME.get(), PORT.get()))
+      server = {
+          'hostname': HOSTNAME.get(),
+          'port': PORT.get()
+      }
+    else:
+      raise PopupException(_('No Sentry servers are configured.'))
+
+  if component:
+    client = client_class(server['hostname'], server['port'], username, component=component)
+  else:
+    client = client_class(server['hostname'], server['port'], username)
+
+  return client
+
+
+def _get_server_properties():
+  try:
+    servers = []
+    sentry_servers = get_sentry_server_rpc_addresses()
+    for server in sentry_servers:
+      servers.append({'hostname': server, 'port': get_sentry_server_rpc_port()})
+  except Exception, e:
+    raise PopupException(_('Error in retrieving Sentry server properties.'), detail=e)
+
+  return servers
 
 
 def _parse_sites():
@@ -107,6 +153,7 @@ def _parse_sites():
   for name, path in paths:
     _SITE_DICT[name] = _parse_site(path)
 
+
 def _parse_site(site_path):
   try:
     data = file(site_path, 'r').read()

+ 107 - 3
desktop/libs/libsentry/src/libsentry/tests.py

@@ -15,24 +15,34 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import lxml.etree
 import os
+import shutil
+import tempfile
 
 from nose.plugins.skip import SkipTest
-from nose.tools import assert_equal, assert_true, assert_false, assert_not_equal
+from nose.tools import assert_equal, assert_true, assert_false, assert_not_equal, assert_raises
 
 from django.contrib.auth.models import User
 
 from desktop.lib.django_test_util import make_logged_in_client
+from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.test_utils import add_to_group, grant_access
+
 from hadoop.pseudo_hdfs4 import is_live_cluster
 
-from libsentry.conf import HOSTNAME, PORT, SENTRY_CONF_DIR
+from libsentry import sentry_site
+from libsentry.api import get_api
+from libsentry.api2 import get_api as get_api2
+from libsentry.conf import is_enabled, HOSTNAME, PORT, SENTRY_CONF_DIR
 from libsentry.client import SentryClient
 
 
 class TestWithSentry:
+
   requires_hadoop = True
 
+
   @classmethod
   def setup_class(cls):
 
@@ -48,10 +58,12 @@ class TestWithSentry:
     grant_access("test", "test", "libsentry")
 
     cls.db = SentryClient(HOSTNAME.get(), PORT.get(), 'test')
+    cls.config_path = os.path.join(SENTRY_CONF_DIR.get(), 'sentry-site.xml')
+
 
   @classmethod
   def teardown_class(cls):
-    pass
+    sentry_site.reset()
 
 
   def test_get_collections(self):
@@ -61,3 +73,95 @@ class TestWithSentry:
 
     resp = self.db.list_sentry_roles_by_group(groupName='*')
     assert_equal(0, resp.status.value, resp)
+
+
+  def test_ha_failover(self):
+    try:
+      tmpdir = tempfile.mkdtemp()
+      finish = SENTRY_CONF_DIR.set_for_testing(tmpdir)
+      rpc_addresses = ','.join(sentry_site.get_sentry_server_rpc_addresses())
+
+      # Test with good-host,bad-host-1,bad-host-2
+      xml = self._sentry_site_xml(rpc_addresses='%s,bad-host-1,bad-host-2' % rpc_addresses)
+      file(os.path.join(tmpdir, 'sentry-site.xml'), 'w').write(xml)
+      sentry_site.reset()
+
+      api = get_api(self.user)
+      assert_equal('%s,bad-host-1,bad-host-2' % rpc_addresses, ','.join(sentry_site.get_sentry_server_rpc_addresses()))
+      resp = api.list_sentry_roles_by_group(groupName='*')
+      assert_true(isinstance(resp, list))
+
+      api2 = get_api2(self.user, 'solr')
+      resp = api2.list_sentry_roles_by_group(groupName='*')
+      assert_true(isinstance(resp, list))
+
+      # Test with bad-host-1,bad-host-2,good-host
+      xml = self._sentry_site_xml(rpc_addresses='bad-host-1,bad-host-2,%s' % rpc_addresses)
+      file(os.path.join(tmpdir, 'sentry-site.xml'), 'w').write(xml)
+      sentry_site.reset()
+
+      api = get_api(self.user)
+      assert_equal('bad-host-1,bad-host-2,%s' % rpc_addresses, ','.join(sentry_site.get_sentry_server_rpc_addresses()))
+      resp = api.list_sentry_roles_by_group(groupName='*')
+      assert_true(isinstance(resp, list))
+
+      api2 = get_api2(self.user, 'solr')
+      resp = api2.list_sentry_roles_by_group(groupName='*')
+      assert_true(isinstance(resp, list))
+
+      # Test with bad-host-1,good-host,bad-host-2
+      xml = self._sentry_site_xml(rpc_addresses='bad-host-1,%s,bad-host-2' % rpc_addresses)
+      file(os.path.join(tmpdir, 'sentry-site.xml'), 'w').write(xml)
+      sentry_site.reset()
+
+      api = get_api(self.user)
+      assert_equal('bad-host-1,%s,bad-host-2' % rpc_addresses, ','.join(sentry_site.get_sentry_server_rpc_addresses()))
+      resp = api.list_sentry_roles_by_group(groupName='*')
+      assert_true(isinstance(resp, list))
+
+      api2 = get_api2(self.user, 'solr')
+      resp = api2.list_sentry_roles_by_group(groupName='*')
+      assert_true(isinstance(resp, list))
+
+      # Test with all bad hosts
+      xml = self._sentry_site_xml(rpc_addresses='bad-host-1,bad-host-2')
+      file(os.path.join(tmpdir, 'sentry-site.xml'), 'w').write(xml)
+      sentry_site.reset()
+
+      api = get_api(self.user)
+      assert_equal('bad-host-1,bad-host-2', ','.join(sentry_site.get_sentry_server_rpc_addresses()))
+      assert_raises(PopupException, api.list_sentry_roles_by_group, '*')
+
+      api2 = get_api2(self.user, 'solr')
+      assert_raises(PopupException, api2.list_sentry_roles_by_group, '*')
+
+      # Test with no rpc hosts and fallback to hostname and port
+      xml = self._sentry_site_xml(rpc_addresses='')
+      file(os.path.join(tmpdir, 'sentry-site.xml'), 'w').write(xml)
+      sentry_site.reset()
+
+      api = get_api(self.user)
+      assert_false(sentry_site.is_ha_enabled(), sentry_site.get_sentry_server_rpc_addresses())
+      assert_true(is_enabled() and HOSTNAME.get() and HOSTNAME.get() != 'localhost')
+      resp = api.list_sentry_roles_by_group(groupName='*')
+      assert_true(isinstance(resp, list))
+
+      api2 = get_api2(self.user, 'solr')
+      resp = api2.list_sentry_roles_by_group(groupName='*')
+      assert_true(isinstance(resp, list))
+    finally:
+      sentry_site.reset()
+      finish()
+      shutil.rmtree(tmpdir)
+
+
+  def _sentry_site_xml(self, rpc_addresses):
+    config = lxml.etree.parse(self.config_path)
+    root = config.getroot()
+    properties = config.findall('property')
+    for prop in properties:
+      name = prop.find('name')
+      if name.text == 'sentry.service.client.server.rpc-address':
+        value = prop.find('value')
+        value.text = rpc_addresses
+        return lxml.etree.tostring(root)