Merge pull request #101 from 4dn-dcic/single_txn_indexing

1.2.0 -- Remove deferred queue

.travis.yml

@@ -1,3 +1,4 @@
+dist: trusty
 language: python
 sudo: true
 cache:
@@ -9,8 +10,6 @@ cache:
   - node_modules
 addons:
   apt:
-    config:
-      retries: true
     packages:
     - oracle-java9-set-default
     - bsdtar

base.ini

@@ -4,7 +4,7 @@ sqlalchemy.url = postgresql:///snowflakes
 retry.attempts = 3
 file_upload_bucket = snowflakes-files-dev
 file_upload_profile_name = snowflakes-files-upload
-elasticsearch.server = localhost:9200
+elasticsearch.server = 127.0.0.1:9200
 aws_ip_ranges_path = %(here)s/aws-ip-ranges.json
 download_proxy = https://download.encodeproject.org/
 annotations_path = %(here)s/annotations.json

development.ini

@@ -6,7 +6,6 @@
 [app:app]
 use = config:base.ini#app
 sqlalchemy.url = postgresql://postgres@:5432/postgres?host=/tmp/snovault/pgdata
-snp_search.server = localhost:9200
 load_test_only = true
 create_tables = true
 testing = true
@@ -39,9 +38,6 @@ use = egg:rutter#urlmap
 [composite:indexer]
 use = config:base.ini#indexer
 
-[composite:file_indexer]
-use = config:base.ini#file_indexer
-
 ###
 # wsgi server configuration
 ###

etc/snovault-apache.conf

@@ -26,14 +26,6 @@ LogFormat "%v:%p %h %l %u %t \"%r\" %>s %O \"%{Referer}i\" \"%{User-Agent}i\" %{
     </IfModule>
 </Directory>
 
-<Directory /srv/snowflakes/parts/production-fileindexer>
-    Order deny,allow
-    Allow from all
-    <IfModule access_compat_module>
-        Require all granted
-    </IfModule>
-</Directory>
-
 
 # Specifying process-group and application-group here ensures processes are started on httpd start
 WSGIScriptAlias / /srv/snowflakes/parts/production/wsgi process-group=snowflakes application-group=%{GLOBAL}
@@ -122,4 +114,3 @@ ErrorDocument 403 "Forbidden. HTTPS required for authenticated access."
 #RewriteCond %{HTTP:X-Forwarded-Proto} =http
 #RewriteCond %{HTTP_HOST} ^(www\.encodeproject\.org|test\.encodedcc\.org)$
 #RewriteRule ^ https://%{HTTP_HOST}%{REQUEST_URI} [redirect=permanent,last,qsappend]
-

production.ini.in

@@ -12,9 +12,6 @@ elasticsearch.aws_auth = true
 [composite:indexer]
 use = config:base.ini#indexer
 
-[composite:fileindexer]
-use = config:base.ini#fileindexer
-
 [pipeline:main]
 pipeline =
     config:base.ini#memlimit

setup.py

@@ -64,17 +64,17 @@
 
 setup(
     name='snovault',
-    version='0.10',
+    version='1.2.0',
     description='Snovault Hybrid Object Relational Database Framework',
     long_description=README + '\n\n' + CHANGES,
     packages=find_packages('src'),
     package_dir={'': 'src'},
     include_package_data=True,
     package_data={'':['nginx-dev.conf']},
     zip_safe=False,
-    author='Benjamin Hitz',
-    author_email='hitz@stanford.edu',
-    url='http://github.com/ENCODE-DCC/snovault/',
+    author='Carl Vitzthum',
+    author_email='[email protected].edu',
+    url='http://github.com/4dn-dcic/snovault/',
     license='MIT',
     install_requires=requires,
     tests_require=tests_require,
@@ -114,7 +114,7 @@
         #   3 - Alpha
         #   4 - Beta
         #   5 - Production/Stable
-        'Development Status :: 3 - Alpha',
+        'Development Status :: 4 - Beta',
 
         # Indicate who your project is intended for
         'Intended Audience :: Developers',

src/snovault/__init__.py

@@ -74,8 +74,8 @@ def main(global_config, **local_config):
     settings = global_config
     settings.update(local_config)
 
-    # TODO: move to dcicutils
-    set_logging(settings.get('elasticsearch.server'), settings.get('production'))
+    set_logging(in_prod=settings.get('production'))
+    # set_logging(settings.get('elasticsearch.server'), settings.get('production'))
 
     # TODO - these need to be set for dummy app
     # settings['snovault.jsonld.namespaces'] = json_asset('snovault:schemas/namespaces.json')

src/snovault/app.py

@@ -76,7 +76,8 @@ def configure_engine(settings):
 
 
 def set_postgresql_statement_timeout(engine, timeout=20 * 1000):
-    """ Prevent Postgres waiting indefinitely for a lock.
+    """
+    Prevent Postgres waiting indefinitely for a lock.
     """
     from sqlalchemy import event
     import psycopg2
@@ -99,21 +100,31 @@ def json_from_path(path, default=None):
     return json.load(open(path))
 
 
-def configure_dbsession(config):
+def configure_dbsession(config, clear_data=False):
+    """
+    Create a sqlalchemy engine and a session that uses it, the latter of which
+    is added to the registry. Handle some extra registration
+    """
+    import snovault.storage
+    import zope.sqlalchemy
+    from sqlalchemy import orm
     from snovault import DBSESSION
+    from snovault.storage import Base
+
     settings = config.registry.settings
     DBSession = settings.pop(DBSESSION, None)
-    if DBSession is None:
+
+    # handle creating the database engine separately with indexer_worker
+    if DBSession is None and not settings.get('indexer_worker'):
         engine = configure_engine(settings)
 
+        # useful for test instances where we want to clear the data
+        if clear_data:
+            Base.metadata.drop_all(engine)
+
         if asbool(settings.get('create_tables', False)):
-            from snovault.storage import Base
             Base.metadata.create_all(engine)
 
-        import snovault.storage
-        import zope.sqlalchemy
-        from sqlalchemy import orm
-
         DBSession = orm.scoped_session(orm.sessionmaker(bind=engine))
         zope.sqlalchemy.register(DBSession)
         snovault.storage.register(DBSession)
@@ -210,18 +221,6 @@ def main(global_config, **local_config):
         config.include('snovault.elasticsearch')
         config.include('.search')
 
-    if 'snp_search.server' in config.registry.settings:
-        addresses = aslist(config.registry.settings['snp_search.server'])
-        config.registry['snp_search'] = Elasticsearch(
-            addresses,
-            serializer=PyramidJSONSerializer(json_renderer),
-            connection_class=TimedRequestsHttpConnection,
-            retry_on_timeout=True,
-            timeout=60,
-            maxsize=50
-        )
-        config.include('.region_search')
-        config.include('.peak_indexer')
     config.include(static_resources)
     config.include(changelogs)
 

src/snovault/commands/es_index_data.py

@@ -23,6 +23,7 @@ def main():
     ''' Indexes app data loaded to elasticsearch '''
 
     import argparse
+    from snovault import set_logging
     parser = argparse.ArgumentParser(
         description="Index data in Elastic Search", epilog=EPILOG,
         formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -40,7 +41,9 @@ def main():
     app = get_app(args.config_uri, args.app_name, options)
 
     # Loading app will have configured from config file. Reconfigure here:
-    logging.getLogger('snovault').setLevel(logging.DEBUG)
+    # Use `es_server=app.registry.settings.get('elasticsearch.server')` when ES logging is working
+    set_logging(in_prod=app.registry.settings.get('production'), level=logging.INFO)
+
     return run(app, args.uuid)
 
 

src/snovault/crud_views.py

@@ -236,6 +236,7 @@ def get_linking_items(context, request, render=None):
     result = {
         'status': 'success',
         '@type': ['result'],
+        'display_title': 'Links to %s' % item_uuid,
         'notification' : '%s has %s items linking to it. This may include rev_links if status != deleted' % (item_uuid, len(links)),
         'uuids_linking_to': links
     }

src/snovault/elasticsearch/create_mapping.py

@@ -39,7 +39,6 @@
 import transaction
 import os
 import argparse
-from snovault import set_logging
 import logging
 from timeit import default_timer as timer
 
@@ -387,6 +386,10 @@ def es_mapping(mapping, agg_items_mapping):
                 'type': 'keyword',
                 'include_in_all': False,
             },
+            'max_sid': {
+                'type': 'keyword',
+                'include_in_all': False,
+            },
             'item_type': {
                 'type': 'keyword',
             },
@@ -1130,6 +1133,7 @@ def run(app, collections=None, dry_run=False, check_first=False, skip_indexing=F
 
 
 def main():
+    from snovault import set_logging
     parser = argparse.ArgumentParser(
         description="Create Elasticsearch mapping", epilog=EPILOG,
         formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -1156,15 +1160,11 @@ def main():
 
     args = parser.parse_args()
 
-    #logging.basicConfig()
     app = get_app(args.config_uri, args.app_name)
 
-
     # Loading app will have configured from config file. Reconfigure here:
-    set_logging(app.registry.settings.get('elasticsearch.server'),
-                app.registry.settings.get('production'), level=logging.INFO)
-    #global log
-    #log = structlog.get_logger(__name__)
+    # Use `es_server=app.registry.settings.get('elasticsearch.server')` when ES logging is working
+    set_logging(in_prod=app.registry.settings.get('production'), level=logging.INFO)
 
     uuids = run(app, collections=args.item_type, dry_run=args.dry_run, check_first=args.check_first,
                 skip_indexing=args.skip_indexing, index_diff=args.index_diff, strict=args.strict,

src/snovault/elasticsearch/es_index_listener.py

@@ -44,10 +44,7 @@ def run(testapp, interval=DEFAULT_INTERVAL, dry_run=False, path='/index', update
     )
 
     # Make sure elasticsearch is up before trying to index.
-    if path == '/index_file':
-        return
-    else:
-        es = testapp.app.registry[ELASTIC_SEARCH]
+    es = testapp.app.registry[ELASTIC_SEARCH]
     es.info()
 
     # main listening loop
@@ -170,6 +167,7 @@ def update_status(error=None, result=None, indexed=None, **kw):
     if 'interval' in settings:
         kwargs['interval'] = float(settings['interval'])
 
+    # daemon thread that actually executes `run` method to call /index
     listener = ErrorHandlingThread(target=run, name='listener', kwargs=kwargs)
     listener.daemon = True
     log.debug('starting listener')
@@ -204,6 +202,7 @@ def internal_app(configfile, app_name=None, username=None):
 
 def main():
     import argparse
+    from snovault import set_logging
     parser = argparse.ArgumentParser(
         description="Listen for changes from postgres and index in elasticsearch",
         epilog=EPILOG,
@@ -221,7 +220,7 @@ def main():
         help="Poll interval between notifications")
     parser.add_argument(
         '--path', default='/index',
-        help="Path of indexing view (/index or /index_file)")
+        help="Path of indexing view")
     parser.add_argument('config_uri', help="path to configfile")
     args = parser.parse_args()
 
@@ -234,14 +233,9 @@ def main():
     if args.verbose or args.dry_run:
         level = logging.DEBUG
 
-    set_logging(app.registry.settings.get('elasticsearch.server'),
-                app.registry.settings.get('production'), level=level)
-    #global log
-    #log = structlog.get_logger(__name__)
-
     # Loading app will have configured from config file. Reconfigure here:
-    #logging.getLogger('snovault').setLevel(logging.DEBUG)
-
+    # Use `es_server=app.registry.settings.get('elasticsearch.server')` when ES logging is working
+    set_logging(in_prod=app.registry.settings.get('production'), level=logging.INFO)
     return run(testapp, args.poll_interval, args.dry_run, args.path)
 
 

src/snovault/elasticsearch/esstorage.py

@@ -1,5 +1,4 @@
 import elasticsearch.exceptions
-from snovault.util import get_root_request
 from elasticsearch.helpers import scan
 from elasticsearch_dsl import Search, Q
 from pyramid.threadlocal import get_current_request
@@ -83,6 +82,10 @@ def uuid(self):
     def sid(self):
         return self.source['sid']
 
+    @property
+    def max_sid(self):
+        return self.source['max_sid']
+
     def used_for(self, item):
         alsoProvides(item, ICachedItem)
 
@@ -284,7 +287,7 @@ def get_rev_links(self, model, rel, *item_types):
 
     def get_sids_by_uuids(self, rids):
         """
-        Currently not implemented. Just return an empty dict
+        Currently not implemented for ES. Just return an empty dict
 
         Args:
             rids (list): list of string rids (uuids)
@@ -294,6 +297,12 @@ def get_sids_by_uuids(self, rids):
         """
         return {}
 
+    def get_max_sid(self):
+        """
+        Currently not implemented for ES. Just return None
+        """
+        return None
+
     def purge_uuid(self, rid, item_type=None, registry=None):
         """
         Purge a uuid from the write storage (Elasticsearch)