Output/logging updates, simplification of multiproc command line arguments.

Author: jaddison

CHANGELOG.md

@@ -1,3 +1,9 @@
+#### 2018-09-17 version 0.1.2
+
+* Simplified command arguments for multiprocessing
+* Moved from print statements to pure Python logging-based output
+* 
+
 #### 2018-09-15 version 0.1
 
 * Initial release

esdocs/__init__.py

@@ -26,7 +26,7 @@
 import logging
 
 __appname__ = __package__
-__version__ = "0.1.1"
+__version__ = "0.1.2"
 
 app_version = "{}/{}".format(__appname__, __version__)
 

esdocs/contrib/esdjango/run.py

@@ -34,7 +34,7 @@ def run():
         # in esdocs.contrib.esdjango.apps
         django.setup()
     except ImportError:
-        print("esdocs-django must be run from the root of your Django project (where manage.py lives).")
+        logger.info("esdocs-django must be run from the root of your Django project (where manage.py lives).")
         return
 
     base_run(DjangoController)

esdocs/contrib/esdjango/serializer.py

@@ -50,7 +50,6 @@ def fetch_data(cls, **kwargs):
             if end and queryset_end > end:
                 queryset_end = end
 
-            print('.', end='', flush=True)
             n = 0
             for n, row in enumerate(queryset[queryset_start: queryset_end]):
                 yield row

esdocs/controller.py

@@ -86,60 +86,58 @@ def run_operation(self, **options):
             pass
 
     def index_list(self, **options):
-        print("Known, managed indexes{}:".format(" (limited results due to --indexes option)" if self.index_names else ""))
+        logger.info("Known, managed indexes{}:".format(
+            " (limited results due to --indexes option)" if self.index_names else ""))
         for name, index in self.indexes.items():
-            print(" - {}".format(name))
+            logger.info(" - {}".format(name))
 
     def index_init(self, **options):
         for name, index in self.new_indexes.items():
             if not self.client.indices.exists(name):
                 self._index_create(index, name, True)
             else:
-                print("Index '{}' already exists. No change made.".format(index._name))
+                logger.info("Index '{}' already exists. No change made.".format(index._name))
 
     def index_update(self, **options):
         for name, index in self.indexes.items():
             if not self.client.indices.exists(name):
                 self._index_create(self.new_indexes[name], name, True)
             else:
-                # index.close(using=self.using)
                 try:
                     index.save(using=self.using)
-                    print("Updated index mapping for '{}'.".format(name))
+                    logger.info("Updated index mapping for '{}'.".format(name))
                 except elasticsearch.exceptions.RequestError as e:
-                    print(str(e))
-                # index.open(using=self.using)
+                    logger.info(str(e))
 
     def index_rebuild(self, **options):
-        multiproc = options.get('multiproc', False)
-        if multiproc:
+        if options.get('multi') is not None:
             policy = ParallelStreamingPolicy(self.parallel_prep)
         else:
             policy = StreamingPolicy()
 
         for name, index in self.new_indexes.items():
             self._index_create(index, name, set_alias=False)
 
-            print("Updating index settings to be bulk-indexing friendly...")
+            logger.info("Updating index settings to be bulk-indexing friendly...")
             original_settings = index.get_settings(using=self.using).get(index._name, {}).get('settings', {})
             index.put_settings(body={
                 "index.number_of_replicas": 0,
                 "index.refresh_interval": '-1'
             })
 
-            print("Indexing data for '{}'...".format(index._name))
+            logger.info("Indexing data for '{}'...".format(index._name))
 
             for serializer in self._serializers[name]:
-                print(" - processing '{}' documents: ".format(serializer.document.__name__), end='', flush=True)
+                logger.info(" - processing '{}' documents".format(serializer.document.__name__))
                 policy.bulk_operation(serializer, index=index._name, client=self.client, **options)
-                print()
+                logger.info()
 
-            print("Data indexed data for '{}'.".format(index._name))
+            logger.info("Data indexed data for '{}'.".format(index._name))
 
-            print("Force merging index data...")
+            logger.info("Force merging index data...")
             index.forcemerge()
 
-            print("Restoring original/default index settings...")
+            logger.info("Restoring original/default index settings...")
             index.put_settings(body={
                 "index.number_of_replicas": original_settings.get('index', {}).get('number_of_replicas', 1),
                 "index.refresh_interval": original_settings.get('index', {}).get('refresh_interval', '1s')
@@ -153,7 +151,7 @@ def index_rebuild(self, **options):
                     {'add': {'index': index._name, 'alias': name}}
                 ]
             })
-            print("Created alias '{}' for '{}'.".format(name, index._name))
+            logger.info("Created alias '{}' for '{}'.".format(name, index._name))
 
         policy.close()
 
@@ -166,10 +164,10 @@ def index_cleanup(self, **options):
     def _index_create(self, index, alias, set_alias=False):
         if not set_alias:
             index.create(using=self.using)
-            print("Created index '{}', no alias set.".format(index._name))
+            logger.info("Created index '{}', no alias set.".format(index._name))
         else:
             index.aliases(**{alias: {}}).create(using=self.using)
-            print("Created index '{}', aliased to '{}'.".format(index._name, alias))
+            logger.info("Created index '{}', aliased to '{}'.".format(index._name, alias))
 
     def _indexes_delete(self, **options):
         old_indexes = []
@@ -182,9 +180,9 @@ def _indexes_delete(self, **options):
             no_input = options.get('no_input', False)
             if no_input:
                 self.client.indices.delete(",".join(old_indexes))
-                print("Deleting old unaliased indexes:")
+                logger.info("Deleting old unaliased indexes:")
                 for _old in old_indexes:
-                    print(" - deleted index '{}'".format(_old))
+                    logger.info(" - deleted index '{}'".format(_old))
             else:
                 for _old in old_indexes:
                     user_input = 'y' if no_input else ''
@@ -194,4 +192,4 @@ def _indexes_delete(self, **options):
                             break
                     if user_input == 'y':
                         self.client.indices.delete(_old)
-                        print(" - deleted index '{}'".format(_old))
+                        logger.info(" - deleted index '{}'".format(_old))

esdocs/policies.py

@@ -73,7 +73,7 @@ def __init__(self, parallel_prep):
         self.parallel_prep = parallel_prep
 
     def bulk_operation(self, serializer, index, client, **options):
-        procs = options.get('processes')
+        procs = options.get('multi')
         if not procs:
             procs = os.cpu_count()
             procs = procs if procs > 1 else 2

esdocs/serializer.py

@@ -243,5 +243,4 @@ def bulk_operation(cls, index=None, client=None, **options):
             if not ok:
                 action, result = result.popitem()
                 doc_id = '/%s/doc/%s' % (index, result['_id'])
-                print('Failed to %s document %s:' % (action, doc_id,), result)
-                # break
+                logger.warning('Failed to {} document {}: {}'.format(action, doc_id, result))

esdocs/utils.py

@@ -32,12 +32,9 @@ def add_parser_arguments(parser):
                         help="Comma-separate list of index names to target")
     parent.add_argument('--using', action='store', default=None,
                         help="Elasticsearch named connection to use")
-
-    parent.add_argument('--multiproc', action='store_true', default=False,
-                        help="Enable multiple process indexing")
-    parent.add_argument('--numprocs', action='store', type=int, default=None,
-                        help="How many processes to use for indexing; "
-                             "default is CPU core count")
+    parent.add_argument('--multi', nargs='?', const=0, type=int,
+                        help="Enable multiple processes and optionally set number of "
+                             "CPU cores to use (defaults to all cores)")
 
     # hack to get comment args into the main parser;
     # stolen from python argparse source code
@@ -50,15 +47,17 @@ def add_parser_arguments(parser):
         parser._defaults.update(defaults)
 
     sps = parser.add_subparsers(title="commands", dest='action')
-    # this is a hack for a Django 1.11 bug: https://code.djangoproject.com/ticket/29295;
+    # this is a hack for a Django 1.11 bug:
+    # https://code.djangoproject.com/ticket/29295;
     # TODO: remove when we drop support for broken Django versions
     sps._parser_class = argparse.ArgumentParser
 
     p = sps.add_parser('list', help="List indexes", parents=[parent])
     p = sps.add_parser('init', help="Initialize indexes", parents=[parent])
     p = sps.add_parser('update', help="Update indexes", parents=[parent])
     p = sps.add_parser('rebuild', help="Rebuild indexes", parents=[parent])
-    p.add_argument('--cleanup', action="store_true", dest='delete_old_indexes', default=False)
+    p.add_argument('--cleanup', action="store_true", dest='delete_old_indexes',
+                   default=False)
     p = sps.add_parser('cleanup', help="Delete unaliased indexes", parents=[parent])
 
 
@@ -69,7 +68,8 @@ def run(controller_klass=None):
     from . import app_version, logger as parent_logger
 
     parser = argparse.ArgumentParser()
-    parser.add_argument("-v", "--verbose", action="store_true", help="increase output verbosity")
+    parser.add_argument("-v", "--verbose", action="store_true",
+                        help="increase output verbosity")
     parser.add_argument("--version", action="version", version=app_version)
     add_parser_arguments(parser)
     args = parser.parse_args()
@@ -86,12 +86,17 @@ def run(controller_klass=None):
 
     parent_logger.addHandler(logh)
 
-    logger.info('Logging enabled at {} verbosity'.format(logging.getLevelName(logger.getEffectiveLevel())))
-    logger.info('Multi-process indexing: {}'.format('available' if gevent_enabled else 'unavailable (set environment variable ESDOCS_GEVENT=1)'))
+    logger.info('Logging enabled at {} verbosity'.format(
+        logging.getLevelName(logger.getEffectiveLevel())))
 
-    if not gevent_enabled and args.multiproc:
-        logger.error('Multi-process indexing not available unless environment variable ESDOCS_GEVENT=1 is set, stopping...')
+    if not gevent_enabled and args.multi is not None:
+        logger.error('Multi-process indexing not available unless environment '
+                     'variable ESDOCS_GEVENT=1 is set, stopping...')
         return
+    else:
+        logger.info('Multi-process indexing: {}'.format(
+            'available' if gevent_enabled else 'unavailable (set environment '
+                                               'variable ESDOCS_GEVENT=1)'))
 
     register_serializers(os.getenv('ESDOCS_SERIALIZER_MODULES'))
     Serializer.register_hooks(os.getenv('ESDOCS_SERIALIZER_COMPATIBILITY_HOOKS'))