Browse Source

HUE-5304 [indexer] Fix Morplhine indexer to work with upstream Solr

Romain Rigaux 8 years ago
parent
commit
de3a762

+ 8 - 13
desktop/libs/indexer/src/indexer/api3.py

@@ -181,16 +181,13 @@ def importer_submit(request):
         fields += [{"name": unique_key_field, "type": "string"}]
         kwargs['rowid'] = unique_key_field
 
-      try:
+      if not client.exists(index_name):
         client.create_index(
             name=index_name,
             fields=fields,
             unique_key_field=unique_key_field,
             df=df
         )
-      except Exception, e:
-        if not 'already exists' in str(e):
-          raise e
 
       data = request.fs.read(source["path"], 0, MAX_UPLOAD_SIZE)
       client.index(name=index_name, data=data, **kwargs)
@@ -253,15 +250,13 @@ def _index(request, file_format, collection_name, query=None, start_time=None, l
     schema_fields += [{"name": unique_field, "type": "string"}]
 
   client = SolrClient(user=request.user)
-  try:
-    client.get_index_schema(collection_name)
-  except Exception, e:
-    if 'not available' in str(e):
-      client.create_index(
-        name=collection_name,
-        fields=request.POST.get('fields', schema_fields),
-        unique_key_field=unique_field
-      )
+
+  if not client.exists(collection_name):
+    client.create_index(
+      name=collection_name,
+      fields=request.POST.get('fields', schema_fields),
+      unique_key_field=unique_field
+    )
 
   if file_format['inputFormat'] == 'table':
     db = dbms.get(request.user)

+ 2 - 1
desktop/libs/indexer/src/indexer/conf.py

@@ -29,7 +29,8 @@ from libzookeeper import conf as libzookeeper_conf
 LOG = logging.getLogger(__name__)
 
 
-def get_solr_ensemble():  
+# Deprecated. Should be automatically guessed from Solr admin info API now.
+def get_solr_ensemble():
   return '%s%s' % (libzookeeper_conf.ENSEMBLE.get(), libsolr_conf.SOLR_ZK_PATH.get())
 
 

+ 7 - 5
desktop/libs/indexer/src/indexer/indexers/morphline.py

@@ -25,7 +25,6 @@ from django.utils.translation import ugettext as _
 from mako.lookup import TemplateLookup
 
 from desktop.models import Document2
-from libsolr.conf import SOLR_ZK_PATH
 from libzookeeper.conf import ENSEMBLE
 from notebook.connectors.base import get_api
 from notebook.models import Notebook, make_notebook
@@ -35,6 +34,7 @@ from indexer.conf import CONFIG_INDEXER_LIBS_PATH
 from indexer.fields import get_field_type
 from indexer.file_format import get_file_format_instance, get_file_format_class
 from indexer.indexers.morphline_operations import get_checked_args
+from indexer.solr_client import SolrClient
 
 
 LOG = logging.getLogger(__name__)
@@ -69,7 +69,7 @@ class MorphlineIndexer(object):
     workspace_path = self._upload_workspace(morphline)
 
     task = make_notebook(
-      name=_('Indexing into %s %s') % (collection_name, input_path),
+      name=_('Indexing into %s') % collection_name,
       editor_type='notebook',
       on_success_url=reverse('search:browse', kwargs={'name': collection_name}),
       is_task=True,
@@ -91,6 +91,8 @@ class MorphlineIndexer(object):
 
       task.add_hive_snippet(snippet['database'], sql)
 
+    client = SolrClient(self.user)
+
     task.add_java_snippet(
       clazz='org.apache.solr.hadoop.MapReduceIndexerTool',
       app_jar=lib_path if lib_path is not None else CONFIG_INDEXER_LIBS_PATH.get(),
@@ -103,7 +105,7 @@ class MorphlineIndexer(object):
           u'log4j.properties',
           u'--go-live',
           u'--zk-host',
-          ENSEMBLE.get() + SOLR_ZK_PATH.get(),
+          client.get_zookeeper_host(),
           u'--collection',
           collection_name,
           input_path,
@@ -144,7 +146,6 @@ class MorphlineIndexer(object):
     return [field for field in self.get_field_list(field_data) if field['keep']]
 
   def get_unique_field(self, format_):
-    # check for a unique field
     unique_fields = [column['name'] for column in format_['columns'] if column['unique']]
 
     if unique_fields:
@@ -172,6 +173,7 @@ class MorphlineIndexer(object):
   def generate_morphline_config(self, collection_name, data, uuid_name=None):
     geolite_loc = os.path.join(CONFIG_INDEXER_LIBS_PATH.get(), "GeoLite2-City.mmdb")
     grok_dicts_loc = os.path.join(CONFIG_INDEXER_LIBS_PATH.get(), "grok_dictionaries")
+    client = SolrClient(self.user)
 
     properties = {
       "collection_name": collection_name,
@@ -184,7 +186,7 @@ class MorphlineIndexer(object):
       "get_kept_args": get_checked_args,
       "grok_dictionaries_location" : grok_dicts_loc if self.fs and self.fs.exists(grok_dicts_loc) else None,
       "geolite_db_location" : geolite_loc if self.fs and self.fs.exists(geolite_loc) else None,
-      "zk_host": ENSEMBLE.get()
+      "zk_host": client.get_zookeeper_host()
     }
 
     oozie_workspace = CONFIG_INDEXING_TEMPLATES_PATH.get()