migration.py

# Copyright 2018 Markus Scheidgen
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an"AS IS" BASIS,
# 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.

"""
This module contains functions to read data from NOMAD coe, external sources,
other/older nomad@FAIRDI instances to mass upload it to a new nomad@FAIRDI instance.

.. autoclass:: NomadCOEMigration
.. autoclass:: SourceCalc
"""

from typing import Generator, Tuple, List, Iterable, Any, Dict, BinaryIO, Callable, ContextManager
import os
import os.path
import zipfile
import math
from mongoengine import Document, IntField, StringField, DictField
import datetime
from bravado.exception import HTTPNotFound, HTTPBadRequest, HTTPGatewayTimeout
import os
import runstats
import io
import threading
from tarfile import TarFile
from contextlib import contextmanager

from nomad import utils, infrastructure, files, config
from nomad.coe_repo import User, Calc, LoginException
from nomad.datamodel import CalcWithMetadata
from nomad.processing import FAILURE, SUCCESS


default_pid_prefix = 7000000
""" The default pid prefix for new non migrated calculations """

max_package_size = 16 * 1024 * 1024 * 1024  # 16 GB
""" The maximum size of a package that will be used as an upload on nomad@FAIRDI """
use_stats_for_filestats_threshold = 1024

default_comment = 'entry with unknown provernance'
default_uploader = dict(id=1)


def iterable_to_stream(iterable, buffer_size=io.DEFAULT_BUFFER_SIZE):
    """
    Lets you use an iterable (e.g. a generator) that yields bytestrings as a read-only
    input stream.

    The stream implements Python 3's newer I/O API (available in Python 2's io module).
    For efficiency, the stream is buffered.
    """
    class IterStream(io.RawIOBase):
        def __init__(self):
            self.leftover = None
            self.iterator = iter(iterable)

        def readable(self):
            return True

        def readinto(self, b):
            requested_len = len(b)  # We're supposed to return at most this much
            while True:
                try:
                    chunk = next(self.iterator)
                except StopIteration:
                    if len(self.leftover) == 0:
                        return 0  # indicate EOF
                    chunk = self.leftover
                output, self.leftover = chunk[:requested_len], chunk[requested_len:]
                len_output = len(output)
                if len_output == 0:
                    continue  # do not prematurely indicate EOF
                b[:len_output] = output
                return len_output

    return io.BufferedReader(IterStream(), buffer_size=buffer_size)


OpenFunction = Callable[[str], ContextManager[BinaryIO]]
Directory = Tuple[List[str], int]


class Package(Document):
    """
    A Package represents split origin NOMAD CoE uploads. We use packages as uploads
    in nomad@FAIRDI. Some of the uploads in nomad are very big (alfow lib) and need
    to be split down to yield practical (i.e. for mirrors) upload sizes. Therefore,
    uploads are split over multiple packages if one upload gets to large. A package
    always contains full directories of files to preserve *mainfile* *aux* file relations.
    Package have a package entry in mongo and a .zip file with the raw data.
    """

    package_id = StringField(primary_key=True)
    """ A random UUID for the package. Could serve later its target upload id."""
    package_path = StringField(required=True)
    """ The path to the package .zip file. """
    upload_path = StringField(required=True)
    """ The absolute path of the source upload """
    upload_id = StringField(required=True)
    """ The source upload_id. There might be multiple packages per upload (this is the point). """
    restricted = IntField(default=0)
    """ The restricted in month, 0 for unrestricted. """
    size = IntField()
    """ The sum of all file sizes. """
    files = IntField()
    """ The number of files. """

    migration_version = IntField(default=-1)
    """ The version of the last successful migration of this package """
    report = DictField()
    """ The report of the last successful migration of this package """

    meta = dict(indexes=['upload_id', 'migration_version'])

    @classmethod
    def aggregate_reports(cls, migration_version: str = None) -> 'Report':
        """
        Returns an aggregated report over all package reports of the given migration version,
        or all packages.
        """
        if migration_version is not None:
            query = cls.objects(migration_version__gte=migration_version, report__exists=True)
        else:
            query = cls.objects(report__exists=True)

        report = Report()
        for package in query:
            report.add(Report(package.report))

        return report

    @classmethod
    def create_packages(cls, upload_path: str, target_dir: str) -> Iterable['Package']:
        """
        Will create packages for the given upload_path. Creates the package zip files and
        package index entries. Either will only be created if it does not already exist.
        Yields the Package objects.
        """
        upload_id = os.path.basename(upload_path)
        logger = utils.get_logger(__name__, source_upload_path=upload_path, source_upload_id=upload_id)

        if not os.path.isdir(upload_path):
            logger.error('upload path is not a directory')
            return []

        package_query = cls.objects(upload_id=upload_id)
        upload_directory = files.DirectoryObject(target_dir, upload_id, create=True, prefix=True)
        restricted = 0

        if package_query.count() == 0:
            def open_package_zip(package_entry: 'Package'):
                return zipfile.ZipFile(package_entry.package_path, 'w')

            def create_package_entry():
                package_id = utils.create_uuid()
                return Package(
                    package_id=package_id,
                    upload_path=upload_path,
                    upload_id=upload_id,
                    package_path=upload_directory.join_file('%s.zip' % package_id).os_path)

            def close_package(package_size: int, package_files: int):
                package_zip.close()
                package_entry.size = package_size
                package_entry.files = package_files
                package_entry.save()

                logger.debug('created package', package_id=package_entry.package_id, size=package_size)

            package_entry = create_package_entry()
            package_size = 0
            package_files = 0
            package_zip = open_package_zip(package_entry)
            for (filepaths, size), open in cls.iterate_directory(upload_path):
                for filepath in filepaths:
                    basepath = os.path.basename(filepath)
                    if basepath.startswith('RESTRICTED'):
                        restricted = 36
                        try:
                            restricted = min(36, int(basepath[len('RESTRICTED_'):]))
                        except Exception:
                            pass

                    with package_zip.open(filepath, "w") as target:
                        with open(filepath) as source:
                            buffer = source.read(1024 * 1024)
                            while len(buffer) > 0:
                                target.write(buffer)
                                buffer = source.read(1024 * 1024)
                    package_files += 1

                if size > max_package_size:
                    logger.warn(
                        'directory exceeds max package size',
                        package_id=package_entry.package_id, size=package_size)

                package_size += size
                if package_size > max_package_size:
                    close_package(package_size, package_files)
                    package_size, package_files = 0, 0
                    package_entry = create_package_entry()
                    package_zip = open_package_zip(package_entry)

            if package_files > 0:
                close_package(package_size, package_files)

            package_query = cls.objects(upload_id=upload_id)
            package_query.update(restricted=restricted)
            logger.debug(
                'packaged upload', source_upload_id=upload_id, source_upload_path=upload_path,
                restricted=restricted)

            return package_query
        else:
            return package_query

    @classmethod
    def iterate_directory(cls, upload_path: str) -> Generator[Tuple[Directory, OpenFunction], None, None]:
        """
        Traverses the given upload path and provides directory and file information.

        Returns:
            A tuple with an open function and directory generator. The open function
            take a path as argument and create a context manager with a BinaryIO file like.
            The path are as provided by the directory generator. The directory generator
            yields directories, which are tuples of a path list and a approx. directory
            size in bytes.
        """
        potential_archive_path = os.path.join(upload_path, 'archive.tar.gz')
        if os.path.isfile(potential_archive_path):
            with TarFile.open(potential_archive_path) as tar_file:
                @contextmanager
                def open_function(filepath):
                    file = tar_file.extractfile(filepath)
                    yield file
                    file.close()

                for dir in cls._iterate_upload_archive(tar_file):
                    yield dir, open_function
        else:
            def open_function(filepath):
                return open(os.path.join(upload_path, filepath), 'rb')

            for dir in cls._iterate_upload_directory(upload_path):
                yield dir, open_function

    @classmethod
    def _iterate_upload_directory(cls, upload_path) -> Generator[Directory, None, None]:
        """
        Interprets the given upload path as a directory. Files path are given as upload
        path relative paths.
        """
        stats = runstats.Statistics()
        for root, _, files in os.walk(upload_path):
            directory_filepaths: List[str] = []
            directory_size = 0

            if len(files) == 0:
                continue

            for file in files:
                filepath = os.path.join(root, file)
                filename = filepath[len(upload_path) + 1:]
                directory_filepaths.append(filename)
                # getting file stats is pretty expensive with gpfs
                # if an upload has more then 1000 files, its pretty likely that
                # size patterns repeat ... goood enough
                if len(stats) < use_stats_for_filestats_threshold:
                    filesize = os.path.getsize(filepath)
                    stats.push(filesize)
                else:
                    filesize = stats.mean()
                directory_size += filesize

            yield directory_filepaths, directory_size

    @classmethod
    def _iterate_upload_archive(cls, tar_file: TarFile) -> Generator[Directory, None, None]:
        """
        Interprets the given upload path as an archive. File paths are as in the archive.
        """
        current_directory = None
        directory_filepaths: List[str] = []
        directory_size = 0

        for f in tar_file.getmembers():
            if f.isfile():
                directory = os.path.dirname(f.name)
                if current_directory != directory and len(directory_filepaths) > 0:
                    yield directory_filepaths, directory_size
                    directory_filepaths = []
                    directory_size = 0
                    current_directory = directory

                directory_filepaths.append(f.name)
                directory_size += f.size

            f = tar_file.next()

        yield directory_filepaths, directory_size


class SourceCalc(Document):
    """
    Mongo document used as a calculation, upload, and metadata db and index
    build from a given source db. Each :class:`SourceCacl` entry relates
    a pid, mainfile, upload "id" with each other for a corressponding calculation.
    It might alos contain the user metadata. The uploads are "id"ed via the
    specific path segment that identifies an upload on the CoE repo FS(s) without
    any prefixes (e.g. $EXTRACTED, /data/upload, etc.)
    """
    pid = IntField(primary_key=True)
    mainfile = StringField()
    upload = StringField()
    metadata = DictField()

    migration_version = IntField(default=-1)

    extracted_prefix = '$EXTRACTED/'
    sites = ['/data/nomad/extracted/', '/nomad/repository/extracted/']
    prefixes = [extracted_prefix] + sites

    meta = dict(indexes=['upload', 'mainfile', 'migration_version'])

    _dataset_cache: dict = {}

    @staticmethod
    def index(source, drop: bool = False, with_metadata: bool = True, per_query: int = 100) \
            -> Generator[Tuple['SourceCalc', int], None, None]:
        """
        Creates a collection of :class:`SourceCalc` documents that represent source repo
        db entries.

        Arguments:
            source: The source db sql alchemy session
            drop: True to drop and create a new collection, update the existing otherwise,
                default is False.
            with_metadata: True to also grab all metadata and store it, default is True.
            per_query: The implementation tries to grab almost all data with a heavely joined
                query on the CoE snoflake/star shaped schema.
                The query cannot ask for the whole db at once: choose how many calculations
                should be read at a time to optimize for your application.

        Returns:
            yields tuples (:class:`SourceCalc`, #calcs_total[incl. datasets])
        """
        logger = utils.get_logger(__name__)
        if drop:
            SourceCalc.drop_collection()

        last_source_calc = SourceCalc.objects().order_by('-pid').first()
        start_pid = last_source_calc.pid if last_source_calc is not None else 0
        source_query = source.query(Calc)
        total = source_query.count() - SourceCalc.objects.count()

        while True:
            query_timer = utils.timer(logger, 'query source db')
            query_timer.__enter__()  # pylint: disable=E1101
            calcs: Iterable[Calc] = source_query \
                .filter(Calc.coe_calc_id > start_pid) \
                .order_by(Calc.coe_calc_id) \
                .limit(per_query)

            source_calcs = []
            for calc in calcs:
                query_timer.__exit__(None, None, None)  # pylint: disable=E1101
                try:
                    filenames = calc.files
                    if filenames is None or len(filenames) == 0:
                        continue  # dataset case

                    filename = filenames[0]
                    if len(filenames) == 1 and (filename.endswith('.tgz') or filename.endswith('.zip')):
                        continue  # also a dataset, some datasets have a downloadable archive

                    for prefix in SourceCalc.prefixes:
                        filename = filename.replace(prefix, '')
                    segments = [file.strip('\\') for file in filename.split('/')]

                    source_calc = SourceCalc(pid=calc.pid)
                    source_calc.upload = segments[0]
                    source_calc.mainfile = os.path.join(*segments[1:])
                    if with_metadata:
                        source_calc.metadata = calc.to_calc_with_metadata().__dict__
                    source_calcs.append(source_calc)
                    start_pid = source_calc.pid

                    yield source_calc, total
                except Exception as e:
                    logger.error('could not index', pid=calc.pid, exc_info=e)

            if len(source_calcs) == 0:
                break
            else:
                with utils.timer(logger, 'write index'):
                    SourceCalc.objects.insert(source_calcs)


NO_PROCESSED_CALCS = 0
FAILED_PROCESSING = 1
FAILED_PUBLISH = 2


class NomadCOEMigration:
    """
    Drives a migration from the NOMAD coe repository db to nomad@FAIRDI.

    Arguments:
        migration_version: The migration version. Only packages/calculations with
            no migration version or a lower migration version are migrated.
        package_directory: The directory that packages are/get stored in.
        threads: Number of threads to run migration in parallel.
        quiet: Prints stats if not quiet
    """

    default_sites = [
        '/nomad/repository/data/uploads',
        '/nomad/repository/data/extracted',
        '/data/nomad/uploaded/',
        '/data/nomad/extracted/']

    default_pid_prefix = int(1e7)

    archive_filename = 'archive.tar.gz'
    """ The standard name for tarred uploads in the CoE repository. """

    def __init__(
            self,
            migration_version: int = 0,
            package_directory: str = None,
            threads: int = 1, quiet: bool = False) -> None:
        self.logger = utils.get_logger(__name__, migration_version=migration_version)

        self.migration_version = migration_version
        self.package_directory = package_directory if package_directory is not None else config.fs.migration_packages
        self._client = None
        self._threads = threads
        self._quiet = quiet

        self.source = infrastructure.repository_db

    @property
    def client(self):
        if self._client is None:
            from nomad.client import create_client
            self._client = create_client()

        return self._client

    def report(self):
        """ returns an aggregated report over all prior migrated packages """
        return Package.aggregate_reports(migration_version=self.migration_version)

    def copy_users(self):
        """ Copy all users. """
        for source_user in self.source.query(User).all():
            if source_user.user_id <= 2:
                # skip first two users to keep example users
                # they probably are either already the example users, or [root, Evgeny]
                continue

            create_user_payload = dict(
                user_id=source_user.user_id,
                email=source_user.email,
                first_name=source_user.first_name,
                last_name=source_user.last_name,
                password=source_user.password,
                created=source_user.created
            )

            try:
                create_user_payload.update(token=source_user.token)
            except LoginException:
                pass

            if source_user.affiliation is not None:
                create_user_payload.update(affiliation=dict(
                    name=source_user.affiliation.name,
                    address=source_user.affiliation.address))

            try:
                self.client.auth.create_user(payload=create_user_payload).response()
                self.logger.info('copied user', user_id=source_user.user_id)
            except HTTPBadRequest as e:
                self.logger.error('could not create user due to bad data', exc_info=e, user_id=source_user.user_id)

    expected_differences = {
        '0d': 'molecule / cluster',
        '3d': 'bulk',
        '2d': '2d / surface',
        '+u': 'gga'
    }

    def validate(self, repo_calc: dict, source_calc: CalcWithMetadata, logger) -> bool:
        """
        Validates the given processed calculation, assuming that the data in the given
        source_calc is correct.

        Returns:
            False, if the calculation differs from the source calc.
        """
        keys_to_validate = [
            'atoms', 'basis_set', 'xc_functional', 'system', 'crystal_system',
            'spacegroup', 'code_name', 'code_version']

        def to_comparable_list(list):
            for item in list:
                if isinstance(item, dict):
                    for key in item.keys():
                        if key.endswith('id'):
                            yield item.get(key)
                else:
                    yield item

        is_valid = True
        for key, target_value in repo_calc.items():
            if key not in keys_to_validate:
                continue

            source_value = getattr(source_calc, key, None)

            def check_mismatch() -> bool:
                # some exceptions
                if source_value in NomadCOEMigration.expected_differences and \
                        target_value == NomadCOEMigration.expected_differences.get(source_value):
                    return True

                logger.info(
                    'source target missmatch', quantity=key,
                    source_value=source_value, target_value=target_value,
                    value_diff='%s->%s' % (str(source_value), str(target_value)))
                return False

            if source_value is None and target_value is not None:
                continue

            if target_value is None and source_value is not None:
                is_valid &= check_mismatch()

            if isinstance(target_value, list):
                source_list = list(to_comparable_list(source_value))
                target_list = list(to_comparable_list(target_value))
                if len(set(source_list).intersection(target_list)) != len(target_list):
                    is_valid &= check_mismatch()
                continue

            if isinstance(source_value, str):
                source_value = source_value.lower()
                target_value = str(target_value).lower()

            if source_value != target_value:
                is_valid &= check_mismatch()

        return is_valid

    def surrogate_metadata(self, source: CalcWithMetadata):
        """
        Compute metadata from the given metadata that can be used for new calcs of the
        same upload.
        """
        return CalcWithMetadata(
            uploader=source.uploader,
            with_embargo=source.with_embargo,
            upload_time=source.upload_time,
            coauthors=source.coauthors,
            shared_with=source.shared_with,
            comment=source.comment,
            references=source.references,
            datasets=source.datasets)

    def set_pid_prefix(self, prefix: int = default_pid_prefix):
        """
        Sets the repo db pid counter to the given values. Allows to create new calcs
        without interfering with migration calcs with already existing PIDs.
        """
        self.logger.info('set pid prefix', pid_prefix=prefix)
        self.client.admin.exec_pidprefix_command(payload=dict(prefix=prefix)).response()

    def call_api(self, operation: str, *args, **kwargs) -> Any:
        """
        Calls nomad via the bravado client. It deals with a very busy nomad and catches,
        backsoff, and retries on gateway timouts. It also circumvents bravados/jsonschemas
        thread safety issues using a global lock on client usage.

        Arguments:
            operation: Comma separated string of api, endpoint, operation,
                e.g. 'uploads.get_upload'.
        """
        op_path = operation.split('.')
        op = self.client
        for op_path_segment in op_path:
            op = getattr(op, op_path_segment)

        sleep = utils.SleepTimeBackoff()
        while True:
            try:
                NomadCOEMigration._client_lock.acquire(blocking=True)
                return op(*args, **kwargs).response().result
            except HTTPGatewayTimeout:
                sleep()
            except Exception as e:
                raise e
            finally:
                NomadCOEMigration._client_lock.release()

    def migrate(self, *args, delete_failed: str = '') -> utils.POPO:
        """
        Migrate the given uploads.

        It takes paths to extracted uploads as arguments.

        Requires :class:`Package` instances for the given upload paths. Those will
        be created, if they do not already exists. The packages determine the uploads
        for the target infrastructure.

        Requires a build :func:`index` to look for existing data in the source db. This
        will be used to add user (and other, PID, ...) metadata and validate calculations.

        Uses PIDs of identified old calculations. Will create new PIDs for previously
        unknown uploads. See :func:`set_pid_prefix` on how to avoid conflicts.

        Arguments:
            upload_path: A filepath to the upload directory.
            delete_failed: String from ``N``, ``U``, ``P`` to determine that uploads with
                no processed calcs (N), failed upload processing (U), or failed publish
                operation (P) should be deleted after the migration attempt.

        Returns: Dictionary with statistics on the migration.
        """

        cv = threading.Condition()
        overall_report = Report()
        threads = []

        def print_report():
            if not self._quiet:
                print(overall_report)

        def migrate_package(package: Package):
            logger = self.logger.bind(
                package_id=package.package_id, source_upload_id=package.upload_id)

            if package.migration_version is not None and package.migration_version >= self.migration_version:
                self.logger.info(
                    'package already migrated, skip it',
                    package_id=package.package_id, source_upload_id=package.upload_id)

                package_report = package.report
                overall_report.skipped_packages += 1
            else:
                try:
                    package_report = self.migrate_package(package, delete_failed=delete_failed)

                except Exception as e:
                    package_report = Report()
                    package_report.failed_packages = 1
                    logger.error(
                        'unexpected exception while migrating packages', exc_info=e)
                finally:
                    package.report = package_report
                    package.migration_version = self.migration_version
                    package.save()

            with cv:
                try:
                    overall_report.add(package_report)

                    migrated_all_packages = all(
                        p.migration_version == self.migration_version
                        for p in Package.objects(upload_id=package.upload_id))

                    if migrated_all_packages:
                        missing_calcs = SourceCalc.objects(
                            upload=package.upload_id, migration_version__ne=self.migration_version).count()
                        total_source_calcs = SourceCalc.objects(upload=package.upload_id).count()

                        overall_report.missing_calcs += missing_calcs
                        overall_report.total_source_calcs += total_source_calcs

                        logger.info('migrated upload')

                    print_report()
                except Exception as e:
                    logger.error('unexpected exception while migrating packages', exc_info=e)

                self._threads += 1
                cv.notify()

        for arg in args:
            for package in Package.create_packages(arg, self.package_directory):
                with cv:
                    cv.wait_for(lambda: self._threads > 0)
                    self._threads -= 1
                    thread = threading.Thread(target=lambda: migrate_package(package))
                    threads.append(thread)
                    thread.start()

        for thread in threads:
            thread.join()

        return overall_report

    _client_lock = threading.Lock()

    def migrate_package(self, package: Package, delete_failed: str = '') -> 'Report':
        """ Migrates the given package. For other params see :func:`migrate`. """

        source_upload_id = package.upload_id
        package_id = package.package_id

        logger = self.logger.bind(package_id=package_id, source_upload_id=source_upload_id)
        logger.debug('start to process package')

        report = Report()
        report.total_packages += 1

        # upload and process the upload file
        with utils.timer(logger, 'upload completed'):
            try:
                upload = self.call_api(
                    'uploads.upload', name=package_id, local_path=package.package_path)
            except Exception as e:
                self.logger.error('could not upload package', exc_info=e)
                report.failed_packages += 1
                return report

        logger = logger.bind(
            source_upload_id=source_upload_id, upload_id=upload.upload_id)

        def delete_upload(reason: int):
            delete = \
                (reason == NO_PROCESSED_CALCS and 'N' in delete_failed) or \
                (reason == FAILED_PROCESSING and 'U' in delete_failed) or \
                (reason == FAILED_PUBLISH and 'P' in delete_failed)

            upload_to_delete = upload

            if delete:
                upload_to_delete = self.call_api(
                    'uploads.delete_upload', upload_id=upload_to_delete.upload_id)

                sleep = utils.SleepTimeBackoff()
                while upload_to_delete.process_running:
                    try:
                        upload_to_delete = self.call_api(
                            'uploads.get_upload', upload_id=upload_to_delete.upload_id)
                        sleep()
                    except HTTPNotFound:
                        # the proc upload will be deleted by the delete operation
                        break
                logger.info('deleted upload after migration failure')
            else:
                logger.warning(
                    'will keep upload after migration failure for debugging',
                    reason=reason, delete_failed=delete_failed)

        # grab source calcs, while waiting for upload
        source_calcs = dict()
        surrogate_source_calc_with_metadata = None
        with utils.timer(logger, 'loaded source metadata'):
            with zipfile.ZipFile(package.package_path) as zf:
                for filenames_chunk in utils.chunks(zf.namelist(), 1000):
                    for source_calc in SourceCalc.objects(
                            upload=source_upload_id, mainfile__in=filenames_chunk):

                        source_calc_with_metadata = CalcWithMetadata(**source_calc.metadata)
                        source_calc_with_metadata.pid = source_calc.pid
                        source_calc_with_metadata.mainfile = source_calc.mainfile
                        source_calcs[source_calc.mainfile] = (source_calc, source_calc_with_metadata)

                        # establish a surrogate for new calcs
                        if surrogate_source_calc_with_metadata is None:
                            surrogate_source_calc_with_metadata = \
                                self.surrogate_metadata(source_calc_with_metadata)

        # try to find a surrogate outside the package, if necessary
        if surrogate_source_calc_with_metadata is None:
            source_calc = SourceCalc.objects(upload=source_upload_id).first()
            if source_calc is not None:
                source_calc_with_metadata = CalcWithMetadata(**source_calc.metadata)
                surrogate_source_calc_with_metadata = \
                    self.surrogate_metadata(source_calc_with_metadata)

        # wait for complete upload
        with utils.timer(logger, 'upload processing completed'):
            sleep = utils.SleepTimeBackoff()
            while upload.tasks_running:
                upload = self.call_api('uploads.get_upload', upload_id=upload.upload_id)
                sleep()

        if upload.tasks_status == FAILURE:
            logger.error('failed to process upload', process_errors=upload.errors)
            report.failed_packages += 1
            delete_upload(FAILED_PROCESSING)
            return report
        else:
            report.total_calcs += upload.calcs.pagination.total

        calc_mainfiles = []
        upload_total_calcs = upload.calcs.pagination.total

        # check for processing errors
        with utils.timer(logger, 'checked upload processing'):
            per_page = 500
            for page in range(1, math.ceil(upload_total_calcs / per_page) + 1):
                upload = self.call_api(
                    'uploads.get_upload', upload_id=upload.upload_id, per_page=per_page,
                    page=page, order_by='mainfile')

                for calc_proc in upload.calcs.results:
                    calc_logger = logger.bind(
                        calc_id=calc_proc.calc_id,
                        mainfile=calc_proc.mainfile)

                    if calc_proc.tasks_status == SUCCESS:
                        calc_mainfiles.append(calc_proc.mainfile)
                    else:
                        report.failed_calcs += 1
                        calc_logger.info(
                            'could not process a calc', process_errors=calc_proc.errors)
                        continue

        # verify upload against source
        calcs_in_search = 0
        with utils.timer(logger, 'verified upload against source calcs'):
            scroll_id = 'first'
            while scroll_id is not None:
                scroll_args: Dict[str, Any] = dict(scroll=True)
                if scroll_id != 'first':
                    scroll_args['scroll_id'] = scroll_id

                search = self.call_api('repo.search', upload_id=upload.upload_id, **scroll_args)

                scroll_id = search.scroll_id

                for calc in search.results:
                    calcs_in_search += 1
                    source_calc, source_calc_with_metadata = source_calcs.get(
                        calc['mainfile'], (None, None))

                    if source_calc is not None:
                        report.migrated_calcs += 1

                        calc_logger = logger.bind(calc_id=calc['calc_id'], mainfile=calc['mainfile'])
                        try:
                            if not self.validate(calc, source_calc_with_metadata, calc_logger):
                                report.calcs_with_diffs += 1
                        except Exception as e:
                            calc_logger.warning('unexpected exception during validation', exc_info=e)
                            report.calcs_with_diffs += 1
                    else:
                        calc_logger.info('processed a calc that has no source')
                        report.new_calcs += 1
                        # guessing the metadata from other calcs in upload/package
                        if surrogate_source_calc_with_metadata is not None:
                            new_calc_with_metadata = CalcWithMetadata(**surrogate_source_calc_with_metadata.to_dict())
                            new_calc_with_metadata.mainfile = calc['mainfile']
                        else:
                            calc_logger.warning('could not determine any metadata for new calc')
                            create_time_epoch = os.path.getctime(package.upload_path)
                            new_calc_with_metadata = CalcWithMetadata(
                                upload_time=datetime.datetime.fromtimestamp(create_time_epoch),
                                with_embargo=package.restricted > 0,
                                comment=default_comment,
                                uploader=default_uploader,
                                mainfile=calc['mainfile'])
                            surrogate_source_calc_with_metadata = new_calc_with_metadata

                        source_calcs[calc['mainfile']] = (None, new_calc_with_metadata)

            if len(calc_mainfiles) != calcs_in_search:
                logger.error('missmatch between processed calcs and calcs found with search')

        # publish upload
        if len(calc_mainfiles) > 0:
            with utils.timer(logger, 'upload published'):
                upload_metadata = dict(with_embargo=(package.restricted > 0))
                upload_metadata['calculations'] = [
                    self._to_api_metadata(source_calc_with_metadata)
                    for _, source_calc_with_metadata in source_calcs.values()]

                upload = self.call_api(
                    'uploads.exec_upload_operation', upload_id=upload.upload_id,
                    payload=dict(operation='publish', metadata=upload_metadata))

                sleep = utils.SleepTimeBackoff()
                while upload.process_running:
                    try:
                        upload = self.call_api('uploads.get_upload', upload_id=upload.upload_id)
                        sleep()
                    except HTTPNotFound:
                        # the proc upload will be deleted by the publish operation
                        break

                if upload.tasks_status == FAILURE:
                    logger.error('could not publish upload', process_errors=upload.errors)
                    report.failed_calcs = report.total_calcs
                    report.migrated_calcs = 0
                    report.calcs_with_diffs = 0
                    report.new_calcs = 0
                    report.failed_packages += 1

                    delete_upload(FAILED_PUBLISH)
                    SourceCalc.objects(upload=source_upload_id, mainfile__in=calc_mainfiles) \
                        .update(migration_version=-1)
                else:
                    SourceCalc.objects(upload=source_upload_id, mainfile__in=calc_mainfiles) \
                        .update(migration_version=self.migration_version)
        else:
            delete_upload(NO_PROCESSED_CALCS)
            logger.info('no successful calcs, skip publish')

        logger.info('migrated package', **report)
        return report

    def _to_api_metadata(self, calc_with_metadata: CalcWithMetadata) -> dict:
        """ Transforms to a dict that fullfils the API's uploade metadata model. """

        return dict(
            _upload_time=calc_with_metadata.upload_time,
            _uploader=calc_with_metadata.uploader['id'],
            _pid=calc_with_metadata.pid,
            references=[ref['value'] for ref in calc_with_metadata.references],
            datasets=[dict(
                id=ds['id'],
                _doi=ds.get('doi', {'value': None})['value'],
                _name=ds.get('name', None)) for ds in calc_with_metadata.datasets],
            mainfile=calc_with_metadata.mainfile,
            with_embargo=calc_with_metadata.with_embargo,
            comment=calc_with_metadata.comment,
            coauthors=list(int(user['id']) for user in calc_with_metadata.coauthors),
            shared_with=list(int(user['id']) for user in calc_with_metadata.shared_with)
        )

    def source_calc_index(self, *args, **kwargs):
        """ see :func:`SourceCalc.index` """
        return SourceCalc.index(self.source, *args, **kwargs)

    def package_index(self, *upload_paths) -> None:
        """
        Creates Package objects and respective package zip files for the given uploads.
        The given uploads are supposed to be path to the extracted upload directories.
        If the upload is already in the index, it is not recreated.
        """
        logger = utils.get_logger(__name__)

        for upload_path in upload_paths:
            try:
                for package_entry in Package.create_packages(upload_path, self.package_directory):
                    logger.info(
                        'package in index',
                        source_upload_path=upload_path,
                        source_upload_id=package_entry.upload_id,
                        package_id=package_entry.package_id)
            except Exception as e:
                logger.error(
                    'could create package from upload',
                    upload_path=upload_path, exc_info=e)
                continue


class Report(utils.POPO):
    def __init__(self, *args, **kwargs):
        self.total_packages = 0
        self.failed_packages = 0
        self.skipped_packages = 0
        self.total_calcs = 0  # the calcs that have been found by the target
        self.total_source_calcs = 0  # the calcs in the source index
        self.failed_calcs = 0  # the calcs found b the target that could not be processed/published
        self.migrated_calcs = 0   # the calcs from the source, successfully added to the target
        self.calcs_with_diffs = 0  # the calcs from the source, successfully added to the target with different metadata
        self.new_calcs = 0  # the calcs successfully added to the target that were not found in the source
        self.missing_calcs = 0  # the calcs in the source, that could not be added to the target due to failure or not founding the calc

        super().__init__(*args, **kwargs)

    def add(self, other: 'Report') -> None:
        for key, value in other.items():
            self[key] = self.get(key, 0) + value

    def __str__(self):
        return (
            'packages: {:,}, skipped: {:,}, source calcs: {:,}, migrated: {:,}, '
            'failed: {:,}, missing: {:,}, new: {:,}'.format(
                self.total_packages, self.skipped_packages,
                self.total_source_calcs, self.migrated_calcs,
                self.failed_calcs, self.missing_calcs,
                self.new_calcs))