浏览代码

HUE-8749 [catalog] Basic Atlas search with text query working

Romain Rigaux 6 年之前
父节点
当前提交
e83788547e

+ 12 - 0
desktop/conf.dist/hue.ini

@@ -1921,6 +1921,18 @@
     # Allow admins to upload the last N executed queries in the quick start wizard. Use 0 to disable.
     ## query_history_upload_limit=10000
 
+  [[catalog]]
+    # The type of Catalog: Apache Atlas, Cloudera Navigator...
+    ## interface=atlas
+    # Catalog API URL (without version suffix).
+    ## api_url=http://localhost:21000/atlas/v2
+
+    # Username of the CM user used for authentication.
+    ## server_user=hue
+    # Password of the user used for authentication.
+    ## server_password=
+
+  # Deprecated by [[catalog]]
   [[navigator]]
     # Navigator API URL (without version suffix).
     ## api_url=http://localhost:7187/api

+ 12 - 0
desktop/conf/pseudo-distributed.ini.tmpl

@@ -1927,6 +1927,18 @@
     # Allow admins to upload the last N executed queries in the quick start wizard. Use 0 to disable.
     ## query_history_upload_limit=10000
 
+  [[catalog]]
+    # The type of Catalog: Apache Atlas, Cloudera Navigator...
+    ## interface=atlas
+    # Catalog API URL (without version suffix).
+    ## api_url=http://localhost:21000/atlas/v2
+
+    # Username of the CM user used for authentication.
+    ## server_user=hue
+    # Password of the user used for authentication.
+    ## server_password=
+
+  # Deprecated by [[catalog]]
   [[navigator]]
     # Navigator API URL (without version suffix).
     ## api_url=http://localhost:7187/api

+ 5 - 5
desktop/core/src/desktop/js/api/apiHelper.js

@@ -49,11 +49,11 @@ const HBASE_API_PREFIX = '/hbase/api/';
 const SAVE_TO_FILE = '/filebrowser/save';
 
 const NAV_URLS = {
-  ADD_TAGS: '/metadata/api/navigator/add_tags',
-  DELETE_TAGS: '/metadata/api/navigator/delete_tags',
-  FIND_ENTITY: '/metadata/api/navigator/find_entity',
-  LIST_TAGS: '/metadata/api/navigator/list_tags',
-  UPDATE_PROPERTIES: '/metadata/api/navigator/update_properties'
+  ADD_TAGS: '/metadata/api/catalog/add_tags',
+  DELETE_TAGS: '/metadata/api/catalog/delete_tags',
+  FIND_ENTITY: '/metadata/api/catalog/find_entity',
+  LIST_TAGS: '/metadata/api/catalog/list_tags',
+  UPDATE_PROPERTIES: '/metadata/api/catalog/update_properties'
 };
 
 const NAV_OPT_URLS = {

+ 57 - 47
desktop/libs/metadata/src/metadata/catalog/atlas_client.py

@@ -27,7 +27,7 @@ from desktop.lib.rest import resource
 from desktop.lib.rest.unsecure_http_client import UnsecureHttpClient
 from desktop.lib.rest.http_client import RestException
 
-from metadata.conf import CATALOG, get_catalog_auth_password, get_catalog_auth_username
+from metadata.conf import CATALOG, get_catalog_auth_password
 from metadata.catalog.base import CatalogAuthException, CatalogApiException, CatalogEntityDoesNotExistException, Api
 
 LOG = logging.getLogger(__name__)
@@ -47,7 +47,7 @@ class AtlasApi(Api):
     super(AtlasApi, self).__init__(user)
 
     self._api_url = CATALOG.API_URL.get().strip('/')
-    self._username = get_catalog_auth_username()
+    self._username = CATALOG.SERVER_USER.get()
     self._password = get_catalog_auth_password()
 
     # Navigator does not support Kerberos authentication while other components usually requires it
@@ -79,9 +79,20 @@ class AtlasApi(Api):
 
   def search_entities_interactive(self, query_s=None, limit=100, offset=0, facetFields=None, facetPrefix=None, facetRanges=None, filterQueries=None, firstClassEntitiesOnly=None, sources=None):
     try:
-      pagination = {
-        'offset': offset,
-        'limit': CATALOG.FETCH_SIZE_SEARCH_INTERACTIVE.get(),
+      query_data = {
+        "excludeDeletedEntities": True,
+        "includeSubClassifications": True,
+        "includeSubTypes": True,
+        "includeClassificationAttributes": True,
+        "entityFilters": None,
+        "tagFilters": None,
+        "attributes": None,
+        "query": "*",
+        "limit": CATALOG.FETCH_SIZE_SEARCH_INTERACTIVE.get(),
+        "offset": offset,
+        "typeName": None,
+        "classification": None,
+        "termName": None
       }
 
       f = {
@@ -191,45 +202,44 @@ class AtlasApi(Api):
       search_terms = [term for term in query_s.strip().split()] if query_s else []
       query = []
       for term in search_terms:
-        if ':' not in term:
-          query.append(self._get_boosted_term(term))
-        else:
-          name, val = term.split(':')
-          if val: # Allow to type non default types, e.g for SQL: type:FIEL*
-            if name == 'type': # Make sure type value still makes sense for the source
-              term = '%s:%s' % (name, val.upper())
-              fq_type = entity_types
-            if name.lower() not in ['type', 'tags', 'owner', 'originalname', 'originaldescription', 'lastmodifiedby']:
-              # User Defined Properties are prefixed with 'up_', i.e. "department:sales" -> "up_department:sales"
-              query.append('up_' + term)
-            else:
-              filterQueries.append(term)
-
-      filterQueries.append('deleted:false')
-
-      body = {'query': ' '.join(query) or '*'}
-      if fq_type:
-        filterQueries += ['{!tag=type} %s' % ' OR '.join(['type:%s' % fq for fq in fq_type])]
-
-      source_ids = self.get_cluster_source_ids()
-      if source_ids:
-        body['query'] = source_ids + '(' + body['query'] + ')'
-
-      body['facetFields'] = facetFields or [] # Currently mandatory in API
-      if facetPrefix:
-        body['facetPrefix'] = facetPrefix
-      if facetRanges:
-        body['facetRanges'] = facetRanges
-      if filterQueries:
-        body['filterQueries'] = filterQueries
-      if firstClassEntitiesOnly:
-        body['firstClassEntitiesOnly'] = firstClassEntitiesOnly
-
-      data = json.dumps(body)
+        query.append(term)
+        # if ':' not in term:
+        #   query.append(self._get_boosted_term(term))
+        # else:
+        #   name, val = term.split(':')
+        #   if val: # Allow to type non default types, e.g for SQL: type:FIEL*
+        #     if name == 'type': # Make sure type value still makes sense for the source
+        #       term = '%s:%s' % (name, val.upper())
+        #       fq_type = entity_types
+        #     if name.lower() not in ['type', 'tags', 'owner', 'originalname', 'originaldescription', 'lastmodifiedby']:
+        #       # User Defined Properties are prefixed with 'up_', i.e. "department:sales" -> "up_department:sales"
+        #       query.append('up_' + term)
+        #     else:
+        #       filterQueries.append(term)
+
+      # filterQueries.append('deleted:false')
+
+      query_data['query'] = ' '.join(query) or '*'
+
+      body = {}
+      # if fq_type:
+      #   filterQueries += ['{!tag=type} %s' % ' OR '.join(['type:%s' % fq for fq in fq_type])]
+
+      # body['facetFields'] = facetFields or [] # Currently mandatory in API
+      # if facetPrefix:
+      #   body['facetPrefix'] = facetPrefix
+      # if facetRanges:
+      #   body['facetRanges'] = facetRanges
+      # if filterQueries:
+      #   body['filterQueries'] = filterQueries
+      # if firstClassEntitiesOnly:
+      #   body['firstClassEntitiesOnly'] = firstClassEntitiesOnly
+
+      data = json.dumps(query_data)
       LOG.info(data)
 
-      response = self._root.get('/search/basic?typeName=hbase_table') #?limit=%(limit)s&offset=%(offset)s' % pagination)
-      response['results'] = [self._massage_entity(entity) for entity in response.pop('entities')]
+      response = self._root.post('/search/basic', data=data, contenttype=_JSON_CONTENT_TYPE)
+      response['results'] = [self._massage_entity(entity) for entity in response.pop('entities', [])]
 
       return response
     except RestException, e:
@@ -244,9 +254,9 @@ class AtlasApi(Api):
     return {
         "name": entity['attributes'].get('name', entity['attributes'].get('qualifiedName')),
         "description": entity['attributes'].get('description'),
-         "owner": entity.get('owner'),
-         "sourceType": entity['typeName'],
-         "partColNames":[
+        "owner": entity.get('owner'),
+        "sourceType": entity['typeName'],
+        "partColNames":[
             # "date"
          ],
          "type": "TABLE", # TODO
@@ -261,9 +271,9 @@ class AtlasApi(Api):
          "properties":{
          },
          "identity": entity['guid'],
-         "created": entity['attributes']['createTime'], #"2019-03-28T19:30:30.000Z",
+         "created": 'createTime' in entity['attributes'] and entity['attributes']['createTime'], #"2019-03-28T19:30:30.000Z",
          "parentPath": "/default",
-         "originalName": entity['attributes']['qualifiedName'],
+         "originalName": entity['attributes'].get('qualifiedName'),
         #  "lastAccessed":"1970-01-01T00:00:00.000Z"
         #  "clusteredByColNames":null,
         #  "outputFormat":"org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat",

+ 18 - 18
desktop/libs/metadata/src/metadata/catalog_api.py

@@ -32,7 +32,7 @@ from desktop.lib.i18n import force_unicode, smart_str
 
 from metadata.catalog.base import get_api
 from metadata.catalog.navigator_client import CatalogApiException, CatalogEntityDoesNotExistException, CatalogAuthException
-from metadata.conf import has_catalog, NAVIGATOR, has_catalog_file_search
+from metadata.conf import has_catalog, CATALOG, has_catalog_file_search
 
 
 LOG = logging.getLogger(__name__)
@@ -81,7 +81,7 @@ def search_entities_interactive(request):
   """
   For search autocomplete.
   """
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   query_s = json.loads(request.POST.get('query_s', ''))
   prefix = request.POST.get('prefix')
   offset = request.POST.get('offset', 0)
@@ -107,7 +107,7 @@ def search_entities_interactive(request):
 
   if response.get('facets'): # Remove empty facets
     for fname, fvalues in response['facets'].items():
-      if NAVIGATOR.APPLY_SENTRY_PERMISSIONS.get():
+      if CATALOG.APPLY_SENTRY_PERMISSIONS.get():
         fvalues = []
       else:
         fvalues = sorted([(k, v) for k, v in fvalues.items() if v > 0], key=lambda n: n[1], reverse=True)
@@ -129,7 +129,7 @@ def search_entities(request):
   """
   For displaying results.
   """
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   query_s = json.loads(request.POST.get('query_s', ''))
   query_s = smart_str(query_s)
 
@@ -219,7 +219,7 @@ def _highlight_tags(record, term):
 
 @error_handler
 def list_tags(request):
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   prefix = request.POST.get('prefix')
   offset = request.POST.get('offset', 0)
   limit = request.POST.get('limit', 25)
@@ -240,7 +240,7 @@ def list_tags(request):
 def find_entity(request):
   response = {'status': -1}
 
-  interface = request.GET.get('interface', 'navigator')
+  interface = request.GET.get('interface', CATALOG.INTERFACE.get())
   entity_type = request.GET.get('type', '')
   database = request.GET.get('database', '')
   table = request.GET.get('table', '')
@@ -288,7 +288,7 @@ def find_entity(request):
 def suggest(request):
   response = {'status': -1}
 
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   prefix = request.POST.get('prefix')
 
   api = get_api(request=request, interface=interface)
@@ -305,7 +305,7 @@ def suggest(request):
 def get_entity(request):
   response = {'status': -1}
 
-  interface = request.GET.get('interface', 'navigator')
+  interface = request.GET.get('interface', CATALOG.INTERFACE.get())
   entity_id = request.GET.get('id')
 
   api = get_api(request=request, interface=interface)
@@ -324,7 +324,7 @@ def get_entity(request):
 @require_POST
 @error_handler
 def add_tags(request):
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   entity_id = json.loads(request.POST.get('id', '""'))
   tags = json.loads(request.POST.get('tags', "[]"))
 
@@ -351,7 +351,7 @@ def add_tags(request):
 @require_POST
 @error_handler
 def delete_tags(request):
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   entity_id = json.loads(request.POST.get('id', '""'))
   tags = json.loads(request.POST.get('tags', '[]'))
 
@@ -378,7 +378,7 @@ def delete_tags(request):
 @require_POST
 @error_handler
 def update_properties(request):
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   entity_id = json.loads(request.POST.get('id', '""'))
   properties = json.loads(request.POST.get('properties', '{}')) # Entity properties
   modified_custom_metadata = json.loads(request.POST.get('modifiedCustomMetadata', '{}')) # Aka "Custom Metadata"
@@ -411,7 +411,7 @@ def update_properties(request):
 def delete_metadata_properties(request):
   response = {'status': -1}
 
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   entity_id = json.loads(request.POST.get('id', '""'))
   keys = json.loads(request.POST.get('keys', '[]'))
 
@@ -438,7 +438,7 @@ def delete_metadata_properties(request):
 def get_lineage(request):
   response = {'status': -1, 'inputs': [], 'source_query': '', 'target_queries': [], 'targets': []}
 
-  interface = request.GET.get('interface', 'navigator')
+  interface = request.GET.get('interface', CATALOG.INTERFACE.get())
   entity_id = request.GET.get('id')
 
   api = get_api(request=request, interface=interface)
@@ -470,7 +470,7 @@ def get_lineage(request):
 
 @error_handler
 def create_namespace(request):
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   namespace = request.POST.get('namespace')
   description = request.POST.get('description')
 
@@ -489,7 +489,7 @@ def create_namespace(request):
 
 @error_handler
 def get_namespace(request):
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   namespace = request.POST.get('namespace')
 
   api = get_api(request=request, interface=interface)
@@ -514,7 +514,7 @@ def create_namespace_property(request):
   "type" : "TEXT",
   "createdDate" : "2018-04-02T22:36:19.001Z"
 }"""
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   namespace = request.POST.get('namespace')
   properties = json.loads(request.POST.get('properties', '{}'))
 
@@ -532,7 +532,7 @@ def map_namespace_property(request):
   namespace: "huecatalog",
   name: "relatedEntities"
   }"""
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
   clazz = request.POST.get('class')
   properties = json.loads(request.POST.get('properties', '[]'))
 
@@ -545,7 +545,7 @@ def map_namespace_property(request):
 
 @error_handler
 def get_model_properties_mapping(request):
-  interface = request.POST.get('interface', 'navigator')
+  interface = request.POST.get('interface', CATALOG.INTERFACE.get())
 
   api = get_api(request=request, interface=interface)
 

+ 22 - 110
desktop/libs/metadata/src/metadata/conf.py

@@ -49,6 +49,10 @@ def default_catalog_config_dir():
   """Get from usual main Hue config directory"""
   return get_config_root()
 
+def default_catalog_interface():
+  """Detect if the configured catalog is Navigator or default to Atlas"""
+  return 'navigator' if default_navigator_url() else 'atlas'
+
 def default_navigator_config_dir():
   """Get from usual main Hue config directory"""
   return get_config_root()
@@ -221,160 +225,68 @@ DEFAULT_PUBLIC_KEY = Config(
 # Data Catalog
 
 def get_catalog_url():
-  return (CATALOG.API_URL.get() and CATALOG.API_URL.get().strip('/')[:-3]) or \
-      (NAVIGATOR.API_URL.get() and NAVIGATOR.API_URL.get().strip('/')[:-3])
+  return (CATALOG.API_URL.get() and CATALOG.API_URL.get().strip('/')[:-3]) or get_navigator_url()
 
 def has_catalog(user):
   from desktop.auth.backend import is_admin
-  return bool(get_catalog_url() and get_navigator_auth_password()) \
+  return ((bool(get_catalog_url() and get_catalog_auth_password())) or has_navigator(user)) \
       and (is_admin(user) or user.has_hue_permission(action="access", app=DJANGO_APPS[0]))
 
-
-def get_catalog_auth_type():
-  return CATALOG.AUTH_TYPE.get().lower()
-
-def get_catalog_auth_username():
-  '''Get the username to authenticate with.'''
-
-  if get_catalog_auth_type() == 'ldap':
-    return CATALOG.AUTH_LDAP_USERNAME.get()
-  elif get_catalog_auth_type() == 'saml':
-    return CATALOG.AUTH_SAML_USERNAME.get()
-  else:
-    return CATALOG.AUTH_CM_USERNAME.get()
-
 def get_catalog_auth_password():
   '''Get the password to authenticate with.'''
   global CATALOG_AUTH_PASSWORD
 
   if CATALOG_AUTH_PASSWORD is None:
     try:
-      if get_catalog_auth_type() == 'ldap':
-        CATALOG_AUTH_PASSWORD = CATALOG.AUTH_LDAP_PASSWORD.get()
-      elif get_catalog_auth_type() == 'saml':
-        CATALOG_AUTH_PASSWORD = CATALOG.AUTH_SAML_PASSWORD.get()
-      else:
-        CATALOG_AUTH_PASSWORD = CATALOG.AUTH_CM_PASSWORD.get()
+      CATALOG_AUTH_PASSWORD = CATALOG.SERVER_PASSWORD.get()
     except CalledProcessError:
       LOG.exception('Could not read Catalog password file, need to restart Hue to re-enable it.')
 
   return CATALOG_AUTH_PASSWORD
 
-def get_catalog_cm_password():
-  '''Get default password from secured file'''
-  return CATALOG.AUTH_CM_PASSWORD_SCRIPT.get()
-
-def get_catalog_ldap_password():
-  '''Get default password from secured file'''
-  return CATALOG.AUTH_LDAP_PASSWORD_SCRIPT.get()
-
-def get_catalog_saml_password():
-  '''Get default password from secured file'''
-  return CATALOG.AUTH_SAML_PASSWORD_SCRIPT.get()
-
-def has_catalog_file_search(user):
-  return has_catalog(user) and CATALOG.ENABLE_FILE_SEARCH.get()
-
 CATALOG = ConfigSection(
   key='catalog',
   help=_t("""Configuration options for Catalog API"""),
   members=dict(
-    interface=Config(
+    INTERFACE=Config(
       key='interface',
       help=_t('Type of Catalog to connect to, e.g. Apache Atlas, Navigator...'),
-      default='atlas'),
+      dynamic_default=default_catalog_interface),
+
     API_URL=Config(
       key='api_url',
       help=_t('Base URL to Catalog API.'),
       dynamic_default=default_catalog_url),
-    AUTH_TYPE=Config(
-      key="navmetadataserver_auth_type",
-      help=_t("Which authentication to use: CM or external via LDAP or SAML."),
-      default='CMDB'),
 
-    AUTH_CM_USERNAME=Config(
+    SERVER_USER=Config(
       key="server_user",
       help=_t("Username of the CM user used for authentication."),
       dynamic_default=get_auth_username),
-    AUTH_CM_PASSWORD=Config(
+    SERVER_PASSWORD=Config(
       key="server_password",
-      help=_t("CM password of the user used for authentication."),
-      private=True,
-      dynamic_default=get_catalog_cm_password),
-    AUTH_CM_PASSWORD_SCRIPT=Config(
-      key="navmetadataserver_cmdb_password_script",
-      help=_t("Execute this script to produce the CM password. This will be used when the plain password is not set."),
+      help=_t("Password of the user used for authentication."),
       private=True,
-      type=coerce_password_from_script,
       default=None),
 
-    AUTH_LDAP_USERNAME=Config(
-      key="navmetadataserver_ldap_user",
-      help=_t("Username of the LDAP user used for authentication."),
-      dynamic_default=get_auth_username),
-    AUTH_LDAP_PASSWORD=Config(
-      key="navmetadataserver_ldap_password",
-      help=_t("LDAP password of the user used for authentication."),
-      private=True,
-      dynamic_default=get_catalog_ldap_password),
-    AUTH_LDAP_PASSWORD_SCRIPT=Config(
-      key="navmetadataserver_ldap_password_script",
-      help=_t("Execute this script to produce the LDAP password. This will be used when the plain password is not set."),
-      private=True,
-      type=coerce_password_from_script,
-      default=None),
-
-    AUTH_SAML_USERNAME=Config(
-      key="navmetadataserver_saml_user",
-      help=_t("Username of the SAML user used for authentication."),
-      dynamic_default=get_auth_username),
-    AUTH_SAML_PASSWORD=Config(
-      key="navmetadataserver_saml_password",
-      help=_t("SAML password of the user used for authentication."),
-      private=True,
-      dynamic_default=get_catalog_saml_password),
-    AUTH_SAML_PASSWORD_SCRIPT=Config(
-      key="navmetadataserver_saml_password_script",
-      help=_t("Execute this script to produce the SAML password. This will be used when the plain password  is not set."),
-      private=True,
-      type=coerce_password_from_script,
-      default=None),
-
-    CONF_DIR = Config(
-      key='conf_dir',
-      help=_t('Catalog configuration directory, where client.properties is located.'),
-      dynamic_default=default_catalog_config_dir
-    ),
-    APPLY_SENTRY_PERMISSIONS = Config(
-      key="apply_sentry_permissions",
-      help=_t("Perform privilege filtering. Default to true automatically if the cluster is secure. Only happening when using Sentry."),
-      dynamic_default=get_security_default,
-      type=coerce_bool
-    ),
-    FETCH_SIZE_SEARCH = Config(
-      key="fetch_size_search",
-      help=_t("Max number of items to fetch in one call in object search."),
-      default=450,
-      type=int
-    ),
     FETCH_SIZE_SEARCH_INTERACTIVE = Config(
       key="fetch_size_search_interactive",
       help=_t("Max number of items to fetch in one call in object search autocomplete."),
-      default=450,
+      default=25,
       type=int
     ),
-
-    ENABLE_FILE_SEARCH = Config(
-      key="enable_file_search",
-      help=_t("Enable to search HDFS, S3 files."),
-      type=coerce_bool,
-      default=False
-    )
   )
 )
 
 # Navigator is deprecated over generic Catalog above
 
+def get_navigator_url():
+  return NAVIGATOR.API_URL.get() and NAVIGATOR.API_URL.get().strip('/')[:-3]
+
+def has_navigator(user):
+  from desktop.auth.backend import is_admin
+  return bool(get_navigator_url() and get_navigator_auth_password()) \
+      and (is_admin(user) or user.has_hue_permission(action="access", app=DJANGO_APPS[0]))
+
 def get_navigator_auth_type():
   return NAVIGATOR.AUTH_TYPE.get().lower()