Преглед изворни кода

HUE-2216 [search] Integrate Solr collections API instead of using solrctl

indexer create and delete collections no longer uses solrctl. libzookeeper/models.py now includes a ZookeeperClient wrapper class around Kazoo. Added tests.
Jenny Kim пре 10 година
родитељ
комит
21bad14

+ 23 - 38
desktop/libs/indexer/src/indexer/controller.py

@@ -20,16 +20,16 @@ import json
 import logging
 import os
 import shutil
-import subprocess
 
 from django.utils.translation import ugettext as _
 
 from desktop.lib.exceptions_renderable import PopupException
 from libsolr.api import SolrApi
 from libzookeeper.conf import ENSEMBLE
+from libzookeeper.models import ZookeeperClient
 from search.conf import SOLR_URL, SECURITY_ENABLED
 
-from indexer.conf import SOLRCTL_PATH, CORE_INSTANCE_DIR
+from indexer.conf import CORE_INSTANCE_DIR
 from indexer.utils import copy_configs, field_values_from_log, field_values_from_separated_file
 
 
@@ -37,15 +37,7 @@ LOG = logging.getLogger(__name__)
 MAX_UPLOAD_SIZE = 100 * 1024 * 1024 # 100 MB
 ALLOWED_FIELD_ATTRIBUTES = set(['name', 'type', 'indexed', 'stored'])
 FLAGS = [('I', 'indexed'), ('T', 'tokenized'), ('S', 'stored')]
-
-
-def get_solrctl_path():
-  solrctl_path = SOLRCTL_PATH.get()
-  if solrctl_path is None:
-    LOG.error("Could not find solrctl executable")
-    raise PopupException(_('Could not find solrctl executable'))
-
-  return solrctl_path
+ZK_SOLR_CONFIG_NAMESPACE = 'configs'
 
 
 def get_solr_ensemble():
@@ -139,31 +131,25 @@ class CollectionManagerController(object):
       # Need to remove path afterwards
       tmp_path, solr_config_path = copy_configs(fields, unique_key_field, df, True)
 
-      # Create instance directory.
-      solrctl_path = get_solrctl_path()
-
-      process = subprocess.Popen([solrctl_path, "--zk", get_solr_ensemble(), "instancedir", "--create", name, solr_config_path],
-                                 stdout=subprocess.PIPE,
-                                 stderr=subprocess.PIPE)
-      status = process.wait()
+      zc = ZookeeperClient(hosts=get_solr_ensemble(), read_only=False)
+      root_node = '%s/%s' % (ZK_SOLR_CONFIG_NAMESPACE, name)
+      config_root_path = '%s/%s' % (solr_config_path, 'conf')
+      try:
+        zc.copy_path(root_node, config_root_path)
+      except Exception, e:
+        zc.delete_path(root_node)
+        raise PopupException(_('Error in copying Solr configurations.'), detail=e)
 
       # Don't want directories laying around
       shutil.rmtree(tmp_path)
 
-      if status != 0:
-        LOG.error("Could not create instance directory.\nOutput: %s\nError: %s" % process.communicate())
-        raise PopupException(_('Could not create instance directory. '
-                               'Check if [libzookeeper]ensemble and [indexer]solrctl_path are correct in Hue config.'))
-
       api = SolrApi(SOLR_URL.get(), self.user, SECURITY_ENABLED.get())
       if not api.create_collection(name):
         # Delete instance directory if we couldn't create a collection.
-        process = subprocess.Popen([solrctl_path, "--zk", get_solr_ensemble(), "instancedir", "--delete", name],
-                                   stdout=subprocess.PIPE,
-                                   stderr=subprocess.PIPE)
-        if process.wait() != 0:
-          LOG.error("Cloud not delete collection.\nOutput: %s\nError: %s" % process.communicate())
-        raise PopupException(_('Could not create collection. Check error logs for more info.'))
+        try:
+          zc.delete_path(root_node)
+        except Exception, e:
+          raise PopupException(_('Error in deleting Solr configurations.'), detail=e)
     else:
       # Non-solrcloud mode
       # Create instance directory locally.
@@ -190,15 +176,14 @@ class CollectionManagerController(object):
 
     if api.remove_collection(name):
       # Delete instance directory.
-      solrctl_path = get_solrctl_path()
-
-      process = subprocess.Popen([solrctl_path, "--zk", get_solr_ensemble(), "instancedir", "--delete", name],
-                                 stdout=subprocess.PIPE,
-                                 stderr=subprocess.PIPE
-                                 )
-      if process.wait() != 0:
-        LOG.error("Cloud not delete instance directory.\nOutput stream: %s\nError stream: %s" % process.communicate())
-        raise PopupException(_('Could not create instance directory. Check error logs for more info.'))
+      try:
+        root_node = '%s/%s' % (ZK_SOLR_CONFIG_NAMESPACE, name)
+        zc = ZookeeperClient(hosts=get_solr_ensemble(), read_only=False)
+        zc.delete_path(root_node)
+      except Exception, e:
+        # Re-create collection so that we don't have an orphan config
+        api.add_collection(name)
+        raise PopupException(_('Error in deleting Solr configurations.'), detail=e)
     else:
       raise PopupException(_('Could not remove collection. Check error logs for more info.'))
 

+ 5 - 3
desktop/libs/libsentry/src/libsentry/api.py

@@ -24,11 +24,10 @@ import time
 from django.utils.translation import ugettext as _
 
 from desktop.lib.exceptions_renderable import PopupException
-from libzookeeper.models import get_children_data
-
 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 libzookeeper.models import ZookeeperClient
 
 
 LOG = logging.getLogger(__name__)
@@ -254,13 +253,16 @@ def _get_server_properties():
       if not _api_cache:
 
         servers = []
-        sentry_servers = get_children_data(ensemble=get_sentry_server_ha_zookeeper_quorum(), namespace=get_sentry_server_ha_zookeeper_namespace())
+        client = ZookeeperClient(hosts=get_sentry_server_ha_zookeeper_quorum())
+        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()
 

+ 74 - 19
desktop/libs/libzookeeper/src/libzookeeper/models.py

@@ -14,37 +14,92 @@
 # 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 os
 
 from kazoo.client import KazooClient
 
 from hadoop import cluster
-from desktop.lib.exceptions_renderable import PopupException
+from libzookeeper.conf import ENSEMBLE, PRINCIPAL_NAME
 
-from libzookeeper.conf import PRINCIPAL_NAME
 
+LOG = logging.getLogger(__name__)
 
-def get_children_data(ensemble, namespace, read_only=True):
-  hdfs = cluster.get_hdfs()
-  if hdfs is None:
-    raise PopupException(_('No [hdfs] configured in hue.ini.'))
 
-  if hdfs.security_enabled:
-    sasl_server_principal = PRINCIPAL_NAME.get()
-  else:
-    sasl_server_principal = None
+class ReadOnlyClientException(Exception):
+  pass
 
-  zk = KazooClient(hosts=ensemble, read_only=read_only, sasl_server_principal=sasl_server_principal)
 
-  zk.start()
+class ZookeeperConfigurationException(Exception):
+  pass
 
-  children_data = []
 
-  children = zk.get_children(namespace)
+class ZookeeperClient(object):
 
-  for node in children:
-    data, stat = zk.get("%s/%s" % (namespace, node))
-    children_data.append(data)
+  def __init__(self, hosts=None, read_only=True):
+    self.hosts = hosts if hosts else ENSEMBLE.get()
+    self.read_only = read_only
 
-  zk.stop()
+    hdfs = cluster.get_hdfs()
 
-  return children_data
+    if hdfs is None:
+      raise ZookeeperConfigurationException('No [hdfs] configured in hue.ini.')
+
+    if hdfs.security_enabled:
+      self.sasl_server_principal = PRINCIPAL_NAME.get()
+    else:
+      self.sasl_server_principal = None
+
+    self.zk = KazooClient(hosts=self.hosts,
+                          read_only=self.read_only,
+                          sasl_server_principal=self.sasl_server_principal)
+
+
+  def get_children_data(self, namespace):
+    children_data = []
+    try:
+      self.zk.start()
+
+      children = self.zk.get_children(namespace)
+
+      for node in children:
+        data, stat = self.zk.get("%s/%s" % (namespace, node))
+        children_data.append(data)
+    finally:
+      self.zk.stop()
+
+    return children_data
+
+
+  def copy_path(self, namespace, filepath):
+    if self.read_only:
+      raise ReadOnlyClientException('Cannot execute copy_path when read_only is set to True.')
+
+    try:
+      self.zk.start()
+
+      self.zk.ensure_path(namespace)
+      for dir, subdirs, files in os.walk(filepath):
+        path = dir.replace(filepath, '').strip('/')
+        if path:
+          node_path = '%s/%s' % (namespace, path)
+          self.zk.create(path=node_path, value='', makepath=True)
+        for filename in files:
+          node_path = '%s/%s/%s' % (namespace, path, filename)
+          with open(os.path.join(dir, filename), 'r') as f:
+            file_content = f.read()
+            self.zk.create(path=node_path, value=file_content, makepath=True)
+    finally:
+      self.zk.stop()
+
+
+  def delete_path(self, namespace):
+    if self.read_only:
+      raise ReadOnlyClientException('Cannot execute delete_path when read_only is set to True.')
+
+    try:
+      self.zk.start()
+
+      self.zk.delete(namespace, recursive=True)
+    finally:
+      self.zk.stop()

+ 59 - 3
desktop/libs/libzookeeper/src/libzookeeper/tests.py

@@ -15,6 +15,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
+import shutil
+import tempfile
+
 from nose.plugins.skip import SkipTest
 from nose.tools import assert_equal, assert_true, assert_false
 
@@ -24,7 +28,7 @@ from desktop.lib.django_test_util import make_logged_in_client
 from desktop.lib.test_utils import add_to_group, grant_access
 from hadoop.pseudo_hdfs4 import is_live_cluster
 
-from libzookeeper.models import get_children_data
+from libzookeeper.models import ZookeeperClient
 from libzookeeper.conf import zkensemble
 
 
@@ -42,5 +46,57 @@ class TestWithZooKeeper:
     add_to_group('test')
     grant_access("test", "test", "libzookeeper")
 
-  def test_get_collections(self):
-    db = get_children_data(zkensemble(), '/solr')
+    # Create a ZKNode namespace
+    cls.namespace = 'TestWithZooKeeper'
+
+    # Create temporary test directory and file with contents
+    cls.local_directory = tempfile.mkdtemp()
+    # Create subdirectory
+    cls.subdir_name = 'subdir'
+    subdir_path = '%s/%s' % (cls.local_directory, cls.subdir_name)
+    os.mkdir(subdir_path, 0755)
+    # Create file
+    cls.filename = 'test.txt'
+    file_path = '%s/%s' % (subdir_path, cls.filename)
+    cls.file_contents = "This is a test"
+    file = open(file_path, 'w+')
+    file.write(cls.file_contents)
+    file.close()
+
+  @classmethod
+  def teardown_class(cls):
+    # Don't want directories laying around
+    shutil.rmtree(cls.local_directory)
+
+  def test_get_children_data(self):
+    client = ZookeeperClient(hosts=zkensemble())
+    db = client.get_children_data(namespace='')
+    assert_true(len(db) > 0)
+
+  def test_copy_and_delete_path(self):
+    root_node = '%s/%s' % (TestWithZooKeeper.namespace, 'test_copy_and_delete_path')
+    client = ZookeeperClient(hosts=zkensemble(), read_only=False)
+
+    # Delete the root_node first just in case it wasn't cleaned up in previous run
+    client.zk.start()
+    client.zk.delete(root_node, recursive=True)
+    client.zk.stop()
+
+    # Test copy_path
+    client.copy_path(root_node, TestWithZooKeeper.local_directory)
+
+    client.zk.start()
+    assert_true(client.zk.exists('%s' % root_node))
+    assert_true(client.zk.exists('%s/%s' % (root_node, TestWithZooKeeper.subdir_name)))
+    assert_true(client.zk.exists('%s/%s/%s' % (root_node, TestWithZooKeeper.subdir_name, TestWithZooKeeper.filename)))
+    contents, stats = client.zk.get('%s/%s/%s' % (root_node, TestWithZooKeeper.subdir_name, TestWithZooKeeper.filename))
+    assert_equal(contents, TestWithZooKeeper.file_contents)
+    client.zk.stop()
+
+    # Test delete_path
+    client.delete_path(root_node)
+
+    client.zk.start()
+    assert_equal(client.zk.exists('%s' % root_node), None)
+    assert_equal(client.zk.exists('%s/%s' % (root_node, TestWithZooKeeper.subdir_name)), None)
+    assert_equal(client.zk.exists('%s/%s/%s' % (root_node, TestWithZooKeeper.subdir_name, TestWithZooKeeper.filename)), None)