Browse Source

[raz] Add support for RAZ HA (#3558)

What changes were proposed in this pull request?

- Add support for RAZ HA in Hue - multiple RAZ servers (min. 2)
- This changes is similar to the current approach for handling HA for services like Knox, Idbroker, Atlas etc. where we switch the host after error handling.

- There are long term improvements possible with caching healthy hosts with lazy switching -- but that requires some scoping for maintaining the cache state which is in the roadmap.

How was this patch tested?

- Updated existing UTs and added new UTs.
- Tested in live cluster setup - both HA and non-HA.
Harsh Gupta 2 years ago
parent
commit
50536705ed

+ 1 - 1
desktop/core/src/desktop/conf.py

@@ -2235,7 +2235,7 @@ RAZ = ConfigSection(
     API_URL=Config(
     API_URL=Config(
         key='api_url',
         key='api_url',
         help=_('Endpoint to contact'),
         help=_('Endpoint to contact'),
-        type=str,
+        type=coerce_string,
         dynamic_default=_get_raz_url,
         dynamic_default=_get_raz_url,
     ),
     ),
     API_AUTHENTICATION=Config(
     API_AUTHENTICATION=Config(

+ 36 - 8
desktop/core/src/desktop/lib/raz/raz_client.py

@@ -44,7 +44,7 @@ LOG = logging.getLogger()
 class RazClient(object):
 class RazClient(object):
 
 
   def __init__(self, raz_url, auth_type, username, service='s3', service_name='cm_s3', cluster_name='myCluster'):
   def __init__(self, raz_url, auth_type, username, service='s3', service_name='cm_s3', cluster_name='myCluster'):
-    self.raz_url = raz_url.strip('/')
+    self.raz_url = raz_url
     self.auth_type = auth_type
     self.auth_type = auth_type
     self.username = username
     self.username = username
     self.service = service
     self.service = service
@@ -93,17 +93,15 @@ class RazClient(object):
       "context": {}
       "context": {}
     }
     }
     request_headers = {"Content-Type": "application/json"}
     request_headers = {"Content-Type": "application/json"}
-    raz_url = "%s/api/authz/%s/access?doAs=%s" % (self.raz_url, self.service, self.username)
 
 
     if self.service == 'adls':
     if self.service == 'adls':
       self._make_adls_request(request_data, method, path, url_params, resource_path)
       self._make_adls_request(request_data, method, path, url_params, resource_path)
     elif self.service in ('s3', 'gs'):
     elif self.service in ('s3', 'gs'):
       self._make_s3_request(request_data, request_headers, method, params, headers, url_params, endpoint, resource_path, data=data)
       self._make_s3_request(request_data, request_headers, method, params, headers, url_params, endpoint, resource_path, data=data)
 
 
-    LOG.debug('Raz url: %s' % raz_url)
     LOG.debug("Sending access check headers: {%s} request_data: {%s}" % (request_headers, request_data))
     LOG.debug("Sending access check headers: {%s} request_data: {%s}" % (request_headers, request_data))
 
 
-    raz_req = self._handle_raz_req(raz_url, request_headers, request_data)
+    raz_req = self._handle_raz_req(self.raz_url, request_headers, request_data)
 
 
     signed_response_result = None
     signed_response_result = None
     signed_response = None
     signed_response = None
@@ -144,17 +142,47 @@ class RazClient(object):
   def _handle_raz_req(self, raz_url, request_headers, request_data):
   def _handle_raz_req(self, raz_url, request_headers, request_data):
     if self.auth_type == 'kerberos':
     if self.auth_type == 'kerberos':
       auth_handler = requests_kerberos.HTTPKerberosAuth(mutual_authentication=requests_kerberos.OPTIONAL)
       auth_handler = requests_kerberos.HTTPKerberosAuth(mutual_authentication=requests_kerberos.OPTIONAL)
-      raz_req = requests.post(raz_url, headers=request_headers, json=request_data, auth=auth_handler, verify=False)
+      raz_response = self._handle_raz_ha(raz_url, headers=request_headers, data=request_data, auth_handler=auth_handler)
+
     elif self.auth_type == 'jwt':
     elif self.auth_type == 'jwt':
       jwt_token = fetch_jwt()
       jwt_token = fetch_jwt()
-
       if jwt_token is None:
       if jwt_token is None:
         raise PopupException('Knox JWT is not available to send to RAZ.')
         raise PopupException('Knox JWT is not available to send to RAZ.')
 
 
       request_headers['Authorization'] = 'Bearer %s' % (jwt_token)
       request_headers['Authorization'] = 'Bearer %s' % (jwt_token)
-      raz_req = requests.post(raz_url, headers=request_headers, json=request_data, verify=False)
+      raz_response = self._handle_raz_ha(raz_url, headers=request_headers, data=request_data)
+
+    if not raz_response:
+      raise PopupException('Failed to receive a valid response from RAZ.')
+
+    return raz_response
+
+
+  def _handle_raz_ha(self, raz_url, headers, data, auth_handler=None, verify=False):
+    raz_urls_list = raz_url.split(',')
+    params = {
+      'headers': headers, 
+      'json': data,
+      'verify': verify
+    }
+
+    if auth_handler:
+      params['auth'] = auth_handler
+
+    raz_response = None
+    for r_url in raz_urls_list:
+      r_url = "%s/api/authz/%s/access?doAs=%s" % (r_url.rstrip('/'), self.service, self.username)
+      LOG.info('Attempting to connect to RAZ URL: %s' % r_url)
+
+      try:
+        raz_response = requests.post(r_url, **params)
+      except Exception as e:
+        if 'Failed to establish a new connection' in str(e):
+          LOG.warning('Raz URL %s is not available.' % r_url)
 
 
-    return raz_req
+      # Check response for None and if response code is successful (200), return the raz response.
+      if (raz_response is not None) and (raz_response.status_code == 200):
+        return raz_response
 
 
 
 
   def _make_adls_request(self, request_data, method, path, url_params, resource_path):
   def _make_adls_request(self, request_data, method, path, url_params, resource_path):

+ 68 - 3
desktop/core/src/desktop/lib/raz/raz_client_test.py

@@ -34,6 +34,7 @@ class RazClientTest(unittest.TestCase):
   def setUp(self):
   def setUp(self):
     self.username = 'gethue'
     self.username = 'gethue'
     self.raz_url = 'https://raz.gethue.com:8080'
     self.raz_url = 'https://raz.gethue.com:8080'
+    self.raz_urls_ha = 'https://raz_host_1.gethue.com:8080/,https://raz_host_2.gethue.com:8080/'
 
 
     self.s3_path = 'https://gethue-test.s3.amazonaws.com/gethue/data/customer.csv'
     self.s3_path = 'https://gethue-test.s3.amazonaws.com/gethue/data/customer.csv'
     self.adls_path = 'https://gethuestorage.dfs.core.windows.net/gethue-container/user/csso_hueuser/customer.csv'
     self.adls_path = 'https://gethuestorage.dfs.core.windows.net/gethue-container/user/csso_hueuser/customer.csv'
@@ -62,6 +63,7 @@ class RazClientTest(unittest.TestCase):
         with patch('desktop.lib.raz.raz_client.uuid.uuid4') as uuid:
         with patch('desktop.lib.raz.raz_client.uuid.uuid4') as uuid:
 
 
           requests_post.return_value = Mock(
           requests_post.return_value = Mock(
+            status_code=200,
             json=Mock(return_value=
             json=Mock(return_value=
             {
             {
               'operResult': {
               'operResult': {
@@ -118,28 +120,48 @@ class RazClientTest(unittest.TestCase):
     with patch('desktop.lib.sdxaas.knox_jwt.requests_kerberos.HTTPKerberosAuth') as HTTPKerberosAuth:
     with patch('desktop.lib.sdxaas.knox_jwt.requests_kerberos.HTTPKerberosAuth') as HTTPKerberosAuth:
       with patch('desktop.lib.raz.raz_client.requests.post') as requests_post:
       with patch('desktop.lib.raz.raz_client.requests.post') as requests_post:
         with patch('desktop.lib.raz.raz_client.fetch_jwt') as fetch_jwt:
         with patch('desktop.lib.raz.raz_client.fetch_jwt') as fetch_jwt:
-
           request_headers = {}
           request_headers = {}
           request_data = Mock()
           request_data = Mock()
 
 
           # When auth type is Kerberos
           # When auth type is Kerberos
           client = RazClient(self.raz_url, 'kerberos', username=self.username, service="adls", service_name="cm_adls", cluster_name="cl1")
           client = RazClient(self.raz_url, 'kerberos', username=self.username, service="adls", service_name="cm_adls", cluster_name="cl1")
+          client._handle_raz_ha = Mock()
+
           raz_req = client._handle_raz_req(self.raz_url, request_headers, request_data)
           raz_req = client._handle_raz_req(self.raz_url, request_headers, request_data)
 
 
+          client._handle_raz_ha.assert_called_with(
+            'https://raz.gethue.com:8080',
+            auth_handler=HTTPKerberosAuth(),
+            data=request_data,
+            headers={}
+          )
           fetch_jwt.assert_not_called()
           fetch_jwt.assert_not_called()
 
 
           # When auth type is JWT
           # When auth type is JWT
           fetch_jwt.return_value = 'test_jwt_token'
           fetch_jwt.return_value = 'test_jwt_token'
 
 
           client = RazClient(self.raz_url, 'jwt', username=self.username, service="adls", service_name="cm_adls", cluster_name="cl1")
           client = RazClient(self.raz_url, 'jwt', username=self.username, service="adls", service_name="cm_adls", cluster_name="cl1")
+          client._handle_raz_ha = Mock()
+
           raz_req = client._handle_raz_req(self.raz_url, request_headers, request_data)
           raz_req = client._handle_raz_req(self.raz_url, request_headers, request_data)
 
 
+          client._handle_raz_ha.assert_called_with(
+            'https://raz.gethue.com:8080',
+            data=request_data,
+            headers={'Authorization': 'Bearer test_jwt_token'}
+          )
           fetch_jwt.assert_called()
           fetch_jwt.assert_called()
 
 
+          # Should raise PopupException when RAZ response is None
+          client = RazClient(self.raz_url, 'jwt', username=self.username, service="adls", service_name="cm_adls", cluster_name="cl1")
+          client._handle_raz_ha = Mock(return_value=None)
+
+          assert_raises(PopupException, client._handle_raz_req, self.raz_url, request_headers, request_data)
+
           # Should raise PopupException when JWT is None
           # Should raise PopupException when JWT is None
-          fetch_jwt.return_value = None 
+          fetch_jwt.return_value = None
+          client._handle_raz_ha = Mock()
 
 
-          # token = RazToken(raz_url='https://raz.gethue.com:8080', auth_type='jwt')
           assert_raises(PopupException, client._handle_raz_req, self.raz_url, request_headers, request_data)
           assert_raises(PopupException, client._handle_raz_req, self.raz_url, request_headers, request_data)
 
 
 
 
@@ -294,6 +316,7 @@ class RazClientTest(unittest.TestCase):
             with patch('desktop.lib.raz.raz_client.uuid.uuid4') as uuid:
             with patch('desktop.lib.raz.raz_client.uuid.uuid4') as uuid:
 
 
               requests_post.return_value = Mock(
               requests_post.return_value = Mock(
+                status_code=200,
                 json=Mock(return_value=
                 json=Mock(return_value=
                   {
                   {
                     'operResult': {
                     'operResult': {
@@ -354,3 +377,45 @@ class RazClientTest(unittest.TestCase):
               )
               )
               assert_true(resp)
               assert_true(resp)
               assert_equal(resp['AWSAccessKeyId'], 'AKIA23E77ZX2HVY76YGL')
               assert_equal(resp['AWSAccessKeyId'], 'AKIA23E77ZX2HVY76YGL')
+
+
+  def test_handle_raz_ha(self):
+    with patch('desktop.lib.sdxaas.knox_jwt.requests_kerberos.HTTPKerberosAuth') as HTTPKerberosAuth:
+      with patch('desktop.lib.raz.raz_client.requests.post') as requests_post:
+        requests_post.return_value = Mock(status_code=200)
+        request_data = Mock()
+
+        # Non-HA mode
+        client = RazClient(self.raz_url, 'kerberos', username=self.username, service="s3", service_name="cm_s3", cluster_name="cl1")
+        raz_response = client._handle_raz_ha(self.raz_url, auth_handler=HTTPKerberosAuth(), data=request_data, headers={})
+
+        requests_post.assert_called_with(
+          'https://raz.gethue.com:8080/api/authz/s3/access?doAs=gethue',
+          auth=HTTPKerberosAuth(), 
+          headers={}, 
+          json=request_data, 
+          verify=False
+        )
+        assert_equal(raz_response.status_code, 200)
+
+        # HA mode - where first URL sends 200 status code
+        client = RazClient(self.raz_urls_ha, 'kerberos', username=self.username, service="s3", service_name="cm_s3", cluster_name="cl1")
+        raz_response = client._handle_raz_ha(self.raz_urls_ha, auth_handler=HTTPKerberosAuth(), data=request_data, headers={})
+
+        requests_post.assert_called_with(
+          'https://raz_host_1.gethue.com:8080/api/authz/s3/access?doAs=gethue',
+          auth=HTTPKerberosAuth(), 
+          headers={}, 
+          json=request_data, 
+          verify=False
+        )
+        assert_equal(raz_response.status_code, 200)
+
+        # When no RAZ URL is healthy
+        requests_post.return_value = Mock(status_code=404)
+
+        client = RazClient(self.raz_urls_ha, 'kerberos', username=self.username, service="s3", service_name="cm_s3", cluster_name="cl1")
+        raz_response = client._handle_raz_ha(self.raz_urls_ha, auth_handler=HTTPKerberosAuth(), data=request_data, headers={})
+
+        assert_equal(raz_response, None)
+