Merge pull request #3874 from alephdata/release/4.0.0

4.0.0 into develop

Author: stchris

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 3.17.0
+current_version = 4.0.0
 tag_name = {new_version}
 commit = True
 tag = True

.github/workflows/build.yml

@@ -34,6 +34,8 @@ jobs:
 
       - name: Install development dependencies
         run: make dev
+        env:
+          PIP_BREAK_SYSTEM_PACKAGES: 1
 
       - name: Check code formatting
         run: make format-check

.github/workflows/pr.yml

@@ -15,6 +15,8 @@ jobs:
 
       - name: Install development dependencies
         run: make dev
+        env:
+          PIP_BREAK_SYSTEM_PACKAGES: 1
 
       - name: Check code formatting
         run: make format-check

Dockerfile

@@ -1,12 +1,12 @@
 FROM ubuntu:20.04
 ENV DEBIAN_FRONTEND noninteractive
 
-# build-essential 
+# build-essential
 RUN apt-get -qq -y update \
     && apt-get -qq --no-install-recommends -y install locales \
     ca-certificates postgresql-client libpq-dev curl jq \
     python3-pip python3-icu python3-psycopg2 \
-    python3-lxml python3-crypto \
+    python3-lxml python3-crypto git \
     && apt-get -qq -y autoremove \
     && apt-get clean \
     && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* \

Makefile

@@ -9,7 +9,8 @@ all: build upgrade web
 
 services:
 	$(COMPOSE) up -d --remove-orphans \
-		postgres elasticsearch ingest-file
+		postgres elasticsearch ingest-file  \
+    redis rabbitmq
 
 shell: services
 	$(APPDOCKER) /bin/bash

aleph.env.tmpl

@@ -93,6 +93,9 @@ ALEPH_OAUTH_SECRET=
 
 # Queue mechanism
 # REDIS_URL=redis://redis:6379/0
+# RABBITMQ_URL=rabbitmq
+# RABBITMQ_USERNAME=guest
+# RABBITMQ_PASSWORD=guest
 
 # Content options
 ALEPH_OCR_DEFAULTS=eng

aleph/__init__.py

@@ -23,6 +23,7 @@
 logging.getLogger("pdfminer").setLevel(logging.WARNING)
 logging.getLogger("httpstream").setLevel(logging.WARNING)
 logging.getLogger("factory").setLevel(logging.WARNING)
+logging.getLogger("pika").setLevel(logging.WARNING)
 
 # Log all SQL statements:
 # logging.getLogger('sqlalchemy.engine').setLevel(log_level)

aleph/authz.py

@@ -26,6 +26,7 @@ class Authz(object):
     def __init__(self, role_id, roles, is_admin=False, token_id=None, expire=None):
         self.id = role_id
         self.logged_in = role_id is not None
+        self.auth_method = None
         self.roles = set(roles)
         self.is_admin = is_admin
         self.token_id = token_id

aleph/core.py

@@ -2,7 +2,6 @@
 from urllib.parse import urlencode, urljoin
 
 from banal import clean_dict
-from elasticsearch import Elasticsearch, TransportError
 from flask import Flask, request
 from flask import url_for as flask_url_for
 from flask_babel import Babel
@@ -11,13 +10,16 @@
 from flask_migrate import Migrate
 from flask_sqlalchemy import SQLAlchemy
 from flask_talisman import Talisman
+from elasticsearch import Elasticsearch, TransportError
+
 from followthemoney import set_model_locale
-from servicelayer import settings as sls
-from servicelayer.archive import init_archive
 from servicelayer.cache import get_redis
+from servicelayer.archive import init_archive
 from servicelayer.extensions import get_extensions
-from servicelayer.logs import LOG_FORMAT_JSON, configure_logging
-from servicelayer.util import backoff, service_retries
+from servicelayer.util import service_retries, backoff
+from servicelayer.logs import configure_logging, LOG_FORMAT_JSON
+from servicelayer.taskqueue import get_rabbitmq_channel
+from servicelayer import settings as sls
 from werkzeug.local import LocalProxy
 from werkzeug.middleware.profiler import ProfilerMiddleware
 
@@ -195,6 +197,7 @@ def get_cache():
 kv = LocalProxy(get_redis)
 cache = LocalProxy(get_cache)
 archive = LocalProxy(get_archive)
+rabbitmq_channel = LocalProxy(get_rabbitmq_channel)
 
 
 def url_for(*a, **kw):

aleph/index/entities.py

@@ -162,12 +162,15 @@ def index_entity(entity, sync=False):
 
 def index_proxy(collection, proxy, sync=False):
     delete_entity(proxy.id, exclude=proxy.schema, sync=False)
-    return index_bulk(collection, [proxy], sync=sync)
+    return index_bulk([proxy], collection=collection, sync=sync)
 
 
-def index_bulk(collection, entities, sync=False):
+def index_bulk(entities, collection=None, sync=False):
     """Index a set of entities."""
-    entities = (format_proxy(p, collection) for p in entities)
+    if collection:
+        entities = (format_proxy(p, collection) for p in entities)
+    else:
+        entities = (format_proxy(*p) for p in entities)
     entities = (e for e in entities if e is not None)
     bulk_actions(entities, sync=sync)
 

aleph/index/util.py

@@ -1,4 +1,4 @@
-import logging
+import structlog
 from pprint import pprint  # noqa
 from banal import ensure_list, is_mapping
 from elasticsearch import TransportError
@@ -9,7 +9,7 @@
 from aleph.core import es
 from aleph.settings import SETTINGS
 
-log = logging.getLogger(__name__)
+log = structlog.get_logger(__name__)
 
 BULK_PAGE = 500
 # cf. https://www.elastic.co/guide/en/elasticsearch/reference/current/search-request-from-size.html  # noqa
@@ -56,6 +56,10 @@
 }
 
 
+class AlephOperationalException(Exception):
+    pass
+
+
 def get_shard_weight(schema):
     if SETTINGS.TESTING:
         return 1
@@ -180,7 +184,7 @@ def query_delete(index, query, sync=False, **kwargs):
                 request_timeout=MAX_REQUEST_TIMEOUT,
                 timeout=MAX_TIMEOUT,
                 scroll_size=SETTINGS.INDEX_DELETE_BY_QUERY_BATCHSIZE,
-                **kwargs
+                **kwargs,
             )
             return
         except TransportError as exc:
@@ -237,8 +241,7 @@ def _check_response(index, res):
     if res.get("status", 0) > 399 and not res.get("acknowledged"):
         error = res.get("error", {}).get("reason")
         log.error("Index [%s] error: %s", index, error)
-        return False
-    return True
+        raise AlephOperationalException(f"Index {index} error: {error}")
 
 
 def rewrite_mapping_safe(pending, existing):
@@ -286,26 +289,42 @@ def configure_index(index, mapping, settings):
             "timeout": MAX_TIMEOUT,
             "master_timeout": MAX_TIMEOUT,
         }
-        config = es.indices.get(index=index).get(index, {})
+        res = es.indices.get(index=index)
+
+        if len(res) != 1:
+            # This case should never occur.
+            log.error("ES response", res=res)
+            raise AlephOperationalException("ES response is empty or ambiguous.")
+
+        # The ES response is an object with items for every requested index. As we only request
+        # a single index, we extract the first and only item from the response. We cannot simply
+        # extract the response data using the index name as the name we use to request the index
+        # may be an alias whereas the response will always contain the actual un-aliased name.
+        config = list(res.values())[0]
+
         settings.get("index").pop("number_of_shards")
+        log.info(
+            f"[{index}] Current settings.", index=index, settings=config.get("settings")
+        )
         if check_settings_changed(settings, config.get("settings")):
-            res = es.indices.close(ignore_unavailable=True, **options)
-            res = es.indices.put_settings(body=settings, **options)
-            if not _check_response(index, res):
-                return False
+            log.info(f"[{index}] Updated settings.", index=index, settings=settings)
+            _check_response(index, es.indices.close(ignore_unavailable=True, **options))
+            _check_response(index, es.indices.put_settings(body=settings, **options))
+        else:
+            log.info(f"[{index}] No changes detected in settings.", index=index)
+        log.info(
+            f"[{index}] Current mappings.", index=index, mappings=config.get("mappings")
+        )
+        log.info(f"[{index}] New mappings.", index=index, mappings=mapping)
         mapping = rewrite_mapping_safe(mapping, config.get("mappings"))
-        res = es.indices.put_mapping(body=mapping, ignore=[400], **options)
-        if not _check_response(index, res):
-            return False
-        res = es.indices.open(**options)
-        return True
+        _check_response(
+            index, es.indices.put_mapping(body=mapping, ignore=[400], **options)
+        )
+        _check_response(index, es.indices.open(**options))
     else:
-        log.info("Creating index: %s...", index)
+        log.info(f"Creating index: {index}...", index=index)
         body = {"settings": settings, "mappings": mapping}
-        res = es.indices.create(index, body=body)
-        if not _check_response(index, res):
-            return False
-        return True
+        _check_response(index, es.indices.create(index, body=body))
 
 
 def index_settings(shards=5, replicas=SETTINGS.INDEX_REPLICAS):

aleph/logic/collections.py

@@ -1,7 +1,6 @@
 import logging
 from datetime import datetime
 from collections import defaultdict
-from servicelayer.jobs import Job
 
 from aleph.core import db, cache
 from aleph.authz import Authz
@@ -14,6 +13,7 @@
 from aleph.logic.notifications import publish, flush_notifications
 from aleph.logic.documents import ingest_flush, MODEL_ORIGIN
 from aleph.logic.aggregator import get_aggregator
+from aleph.util import random_id
 
 log = logging.getLogger(__name__)
 
@@ -127,14 +127,37 @@ def _generate():
             yield proxy
         log.debug("[%s] Indexed %s entities", collection, idx)
 
-    entities_index.index_bulk(collection, _generate(), sync=sync)
+    entities_index.index_bulk(_generate(), collection=collection, sync=sync)
+
+
+def index_aggregator_bulk(entity_ids, skip_errors=False, sync=False):
+    def _generate():
+        idx = 0
+        for collection_id in entity_ids.keys():
+            collection = Collection.by_id(collection_id)
+            if collection:
+                aggregator = get_aggregator(collection)
+                entities = aggregator.iterate(
+                    entity_id=entity_ids[collection_id], skip_errors=skip_errors
+                )
+                for idx, proxy in enumerate(entities, 1):
+                    if idx > 0 and idx % 1000 == 0:
+                        log.debug("[%s] Index: %s...", collection_id, idx)
+                    yield (proxy, collection)
+                log.debug("[%s] Indexed %s entities", collection_id, idx)
+            else:
+                log.debug(
+                    f"Bulk indexing for inexistent collection id: {collection_id}"
+                )
+
+    entities_index.index_bulk(_generate(), sync=sync)
 
 
 def reingest_collection(
     collection, job_id=None, index=False, flush=True, include_ingest=False
 ):
     """Trigger a re-ingest for all documents in the collection."""
-    job_id = job_id or Job.random_id()
+    job_id = job_id or random_id()
     if flush:
         ingest_flush(collection, include_ingest=include_ingest)
     for document in Document.by_collection(collection.id):

aleph/logic/documents.py

@@ -1,13 +1,12 @@
 import os
 import logging
-from servicelayer.jobs import Job
 
 from aleph.core import db, archive
 from aleph.settings import SETTINGS
 from aleph.model import Document
 from aleph.queues import ingest_entity
-from aleph.queues import OP_INGEST
 from aleph.logic.aggregator import get_aggregator, MODEL_ORIGIN
+from aleph.util import random_id
 
 log = logging.getLogger(__name__)
 
@@ -17,7 +16,7 @@ def ingest_flush(collection, entity_id=None, include_ingest=False):
     aggregator = get_aggregator(collection)
     aggregator.delete(entity_id=entity_id, origin=MODEL_ORIGIN)
     if include_ingest:
-        aggregator.delete(entity_id=entity_id, origin=OP_INGEST)
+        aggregator.delete(entity_id=entity_id, origin=SETTINGS.STAGE_INGEST)
     for stage in SETTINGS.INGEST_PIPELINE:
         aggregator.delete(entity_id=entity_id, origin=stage)
 
@@ -37,7 +36,7 @@ def crawl_directory(collection, path, parent=None, job_id=None):
         # to be consistent with the behaviour of alephclient
         if path.is_dir() and job_id is None:
             document = None
-            job_id = Job.random_id()
+            job_id = random_id()
         else:
             meta = {"file_name": path.name}
             document = Document.save(
@@ -48,7 +47,7 @@ def crawl_directory(collection, path, parent=None, job_id=None):
                 meta=meta,
             )
             db.session.commit()
-            job_id = job_id or Job.random_id()
+            job_id = job_id or random_id()
             proxy = document.to_proxy()
             ingest_flush(collection, entity_id=proxy.id)
             ingest_entity(collection, proxy, job_id=job_id)

aleph/logic/entities.py

@@ -1,4 +1,5 @@
 import logging
+from aleph.settings import SETTINGS
 from banal import ensure_dict
 from pprint import pformat  # noqa
 from flask_babel import gettext
@@ -10,7 +11,6 @@
 from aleph.model import Entity, Document, EntitySetItem, Mapping, Bookmark
 from aleph.index import entities as index
 from aleph.queues import pipeline_entity, queue_task
-from aleph.queues import OP_UPDATE_ENTITY, OP_PRUNE_ENTITY
 from aleph.logic.notifications import flush_notifications
 from aleph.logic.collections import refresh_collection
 from aleph.logic.collections import MODEL_ORIGIN
@@ -47,7 +47,9 @@ def upsert_entity(data, collection, authz=None, sync=False, sign=False, job_id=N
 
     index.index_proxy(collection, proxy, sync=sync)
     refresh_entity(collection, proxy.id)
-    queue_task(collection, OP_UPDATE_ENTITY, job_id=job_id, entity_id=proxy.id)
+    queue_task(
+        collection, SETTINGS.STAGE_UPDATE_ENTITY, job_id=job_id, entity_id=proxy.id
+    )
     return entity.id
 
 
@@ -142,7 +144,9 @@ def delete_entity(collection, entity, sync=False, job_id=None):
     entity_id = collection.ns.sign(entity.get("id"))
     index.delete_entity(entity_id, sync=sync)
     refresh_entity(collection, entity_id)
-    queue_task(collection, OP_PRUNE_ENTITY, job_id=job_id, entity_id=entity_id)
+    queue_task(
+        collection, SETTINGS.STAGE_PRUNE_ENTITY, job_id=job_id, entity_id=entity_id
+    )
 
 
 def prune_entity(collection, entity_id=None, job_id=None):

aleph/logic/notifications.py

@@ -24,7 +24,7 @@
 
 def channel_tag(obj, clazz=None):
     clazz = clazz or type(obj)
-    if clazz == str:
+    if clazz is str:
         return obj
 
     obj = get_entity_id(obj)