medn

Author: psaiz

cernopendata/cold_storage/api.py

@@ -29,6 +29,7 @@
 from datetime import datetime
 
 from flask import current_app
+from flask_mail import Message
 from invenio_db import db
 from invenio_files_rest.models import (
     Bucket,
@@ -37,12 +38,7 @@
     ObjectVersion,
     ObjectVersionTag,
 )
-from invenio_pidstore.models import PersistentIdentifier
 from invenio_records_files.api import FileObject, FilesIterator, Record
-from sqlalchemy import func
-from flask_mail import Message
-
-from cernopendata.api import RecordFilesWithIndex
 
 from .models import RequestMetadata, TransferMetadata
 
@@ -91,7 +87,7 @@ class Transfer:
     @staticmethod
     def create(entry):
         """Create a new transfer entry."""
-        transfer = Transfer(
+        transfer = TransferMetadata(
             action=entry["action"],
             new_filename=entry["new_filename"],
             record_uuid=entry["record_uuid"],
@@ -110,22 +106,23 @@ def create(entry):
     def _get_ongoing_transfers(last_check):
         """Get transfers that need processing."""
         return (
-            Transfer.query.filter(
-                Transfer.last_check <= last_check, Transfer.finished.is_(None)
+            TransferMetadata.query.filter(
+                TransferMetadata.last_check <= last_check,
+                TransferMetadata.finished.is_(None),
             )
-            .order_by(Transfer.last_check)
+            .order_by(TransferMetadata.last_check)
             .all()
         )
 
     @staticmethod
     def is_scheduled(file_id, action):
         """Check if a transfer is already scheduled."""
         return (
-            db.session.query(Transfer.id)
+            db.session.query(TransferMetadata.id)
             .filter(
-                Transfer.action == action,
-                Transfer.file_id == file_id,
-                Transfer.finished.is_(None),
+                TransferMetadata.action == action,
+                TransferMetadata.file_id == file_id,
+                TransferMetadata.finished.is_(None),
             )
             .first()
             is not None
@@ -139,46 +136,6 @@ def _load_class(full_class_path):
         cls = getattr(module, class_name)
         return cls()
 
-    @staticmethod
-    def check_transfers(catalog):
-        """Check all the ongoing transfers."""
-        logger.info("Checking all the ongoing transfers")
-        now = datetime.utcnow()
-        all_status = {}
-        summary = {}
-        for transfer in Transfer._get_ongoing_transfers(now):
-            id = transfer.id
-            transfer.last_check = datetime.utcnow()
-            transfer.status, error = Transfer._load_class(
-                f"{transfer.method}.TransferManager"
-            ).transfer_status(transfer.method_id)
-            all_status[id] = transfer.status
-            if transfer.status not in summary:
-                summary[transfer.status] = 0
-            summary[transfer.status] += 1
-            if transfer.status == "DONE":
-                logger.debug(
-                    f"Transfer {id}: just finished! Let's update the catalog and mark it as done"
-                )
-                transfer.finished = datetime.now()
-                catalog.add_copy(
-                    transfer.record_uuid,
-                    transfer.file_id,
-                    transfer.action,
-                    transfer.new_filename,
-                )
-            if transfer.status == "FAILED" or not transfer.status:
-                logger.error(f"The transfer {id} failed :(")
-                transfer.reason = error
-                transfer.finished = datetime.now()
-            else:
-                logger.debug(f"Transfer {id} is in status {transfer.status}")
-            db.session.add(transfer)
-            db.session.commit()
-        catalog.reindex_entries()
-        logger.info(f"Summary: {summary}")
-        return all_status
-
 
 class Request:
     """Class to check the cold storage requests."""
@@ -247,148 +204,3 @@ def subscribe(transfer_id, email):
             db.session.commit()
             return True
         return False
-
-    def check_requests(self, manager):
-        """Check the active requests."""
-        # The requests would go through these stages
-        # SUBMITTED ->  STARTED -> COMPLETED
-        self._check_submitted(manager)
-
-        # Now, let's look at the ones that 'started'
-        self._check_running()
-
-    @staticmethod
-    def _check_submitted(manager):
-        """Check if there are any new transfers submitted."""
-        for action in ["stage", "archive"]:
-            active_transfers_count = Transfer.query.filter(
-                Transfer.finished.is_(None), Transfer.action == action
-            ).count()
-            threshold = Request.get_active_transfers_threshold(action)
-
-            logger.info(
-                f"Checking if we can {action} more records: active {active_transfers_count}/{threshold}"
-            )
-            submitted = 0
-            limit = threshold - active_transfers_count
-            if limit > 0:
-                transfers = Request.query.filter_by(
-                    status="submitted", action=action
-                ).all()
-
-                for transfer in transfers:
-                    info = manager.doOperation(
-                        action,
-                        transfer.record_id,
-                        limit=limit,
-                        register=True,
-                        force=False,
-                        dry=False,
-                    )
-                    logger.debug(f"Got {info}")
-                    if info:
-                        submitted += len(info)
-                        transfer.num_files += len(info)
-                        transfer.size += sum(item.size for item in info)
-                    transfer.started_at = datetime.utcnow()
-                    if limit == submitted:
-                        logger.info(
-                            f"Reached the threshold of {threshold} transfers. There might be more in this record"
-                            f"({submitted + active_transfers_count}). Let's wait before continuing"
-                        )
-                    else:
-                        transfer.status = "started"
-                    db.session.add(transfer)
-                    limit -= submitted
-                    if limit <= 0:
-                        logger.info("We have submitted enough. Stopping")
-                        break
-            if submitted:
-                logger.info(f"{submitted} transfers have been submitted!")
-            db.session.commit()
-
-    @staticmethod
-    def _check_running():
-        """Check the records that are being archived."""
-        for action in ["stage", "archive"]:
-
-            requests = Request.query.filter_by(status="started", action=action).all()
-            logger.debug(f"Checking the {len(requests)} {action} requests")
-            completed = 0
-            for request in requests:
-                record = RecordFilesWithIndex.get_record(request.record_id)
-
-                if action == "stage":
-                    if record["availability"] != "online":
-                        continue
-                elif action == "archive":
-                    files = (f for index in record.file_indices for f in index["files"])
-                    missing = next(
-                        (
-                            f
-                            for f in files
-                            if "tags" not in f or "uri_cold" not in f["tags"]
-                        ),
-                        None,
-                    )
-                    if missing:
-                        logger.debug(f"The file {missing['key']} is not in tape yet...")
-                        continue
-                completed += 1
-                Request.complete(request)
-            logger.info(f"{completed}/{len(requests)} requests have finished")
-
-    @staticmethod
-    def get_requests(
-        status=None,
-        action=None,
-        record=None,
-        summary=False,
-        sort=None,
-        direction=None,
-        page=None,
-        per_page=None,
-    ):
-        """Get the summary of the requests."""
-        if summary:
-            query = db.session.query(
-                Request.status,
-                Request.action,
-                func.count().label("count"),
-                func.sum(Request.num_files).label("files"),
-                func.sum(Request.size).label("size"),
-            )
-        else:
-            query = Request.query
-
-        if status:
-            status_list = status.split(",")
-            query = query.filter(Request.status.in_(status_list))
-
-        if action:
-            action_list = action.split(",")
-            query = query.filter(Request.action.in_(action_list))
-
-        if record:
-            try:
-                uuid = PersistentIdentifier.get("recid", record).object_uuid
-                query = query.filter_by(record_id=uuid)
-            except Exception:
-                query = query.filter(False)
-        if summary:
-            result = query.group_by(Request.status, Request.action).all()
-
-        if sort:
-            column = getattr(Request, sort, None)
-            if column:
-                if direction == "desc":
-                    query = query.order_by(column.desc())
-                else:
-                    query = query.order_by(column.asc())
-
-        if page:
-            result = query.order_by(Request.created_at.desc()).paginate(
-                page=page, per_page=per_page, error_out=False
-            )
-
-        return result

cernopendata/cold_storage/catalog.py

@@ -64,15 +64,9 @@ def get_files_from_record(self, record, limit=None):
             if "_file_indices" in record:
                 for f in record["_file_indices"]:
                     files += f["files"]
-        if limit:
-            start = 0
-            end = len(files)
-            if limit < 0:
-                start = -limit
-            else:
-                end = limit
-            files = files[start:end]
-            logger.debug(f"And the list of files are: f{files}")
+        if limit and limit < 0:
+            logger.debug(f"Skipping the first {limit} files")
+            files = files[-limit:]
         return files
 
     def clear_hot(self, record, file_id):

cernopendata/cold_storage/cli.py

@@ -24,10 +24,6 @@
 
 """Cold Storage CLI."""
 
-import json
-import logging.config
-import sys
-from functools import wraps
 from math import log2
 
 import click
@@ -37,6 +33,7 @@
 from invenio_pidstore.models import PersistentIdentifier
 
 from .manager import ColdStorageManager
+from .service import RequestService, TransferService
 
 argument_record = click.argument("record", nargs=-1, required=True, metavar="RECORD")
 
@@ -47,7 +44,6 @@
     + "with the same file and checksum, import it without issuing the transfer",
 )
 option_dry = click.option("--dry/--do-it", default=False, help="Do not issue transfers")
-option_debug = click.option("--debug/--no-debug", default=False)
 option_force = click.option(
     "--force/--no-force",
     default=False,
@@ -80,16 +76,15 @@ def cold():
 @with_appcontext
 @argument_record
 @option_register
-@option_debug
 @option_limit
 @option_force
 @option_dry
-def archive(record, register, debug, limit, force, dry):
+def archive(record, register, limit, force, dry):
     """Move a record to cold."""
-    _doOperation("archive", record, register, debug, limit, force, dry)
+    _doOperation("archive", record, register, limit, force, dry)
 
 
-def _doOperation(operation, record, register, debug, limit, force, dry):
+def _doOperation(operation, record, register, limit, force, dry):
     """Internal function to do the CLI commands."""
     m = ColdStorageManager(current_app)
     counter = 0
@@ -113,13 +108,12 @@ def _doOperation(operation, record, register, debug, limit, force, dry):
 @with_appcontext
 @argument_record
 @option_register
-@option_debug
 @option_limit
 @option_force
 @option_dry
-def stage(record, register, debug, limit, force, dry):
+def stage(record, register, limit, force, dry):
     """Move a record from cold."""
-    _doOperation("stage", record, register, debug, limit, force, dry)
+    _doOperation("stage", record, register, limit, force, dry)
 
 
 @cold.command()
@@ -133,8 +127,7 @@ def settings():
 @cold.command()
 @with_appcontext
 @argument_record
-@option_debug
-def list(record, debug):
+def list(record):
     """Print the urls for an entry.
 
     By default, it prints the urls for all the files of the entry.
@@ -156,8 +149,6 @@ def list(record, debug):
             click.secho(f"The record '{r}' does not exist.", fg="red")
             continue
         info = m.list(uuid)
-        if debug:
-            print("Printing debug info", info)
         if not info:
             click.secho(f"The record {r} does not exist!")
             stats["errors"] += [r]
@@ -191,25 +182,21 @@ def list(record, debug):
 @argument_record
 @option_limit
 @option_dry
-@option_debug
-def clear_hot(record, limit, dry, debug):
+def clear_hot(record, limit, dry):
     """Delete the hot copy of a file that has a cold copy."""
-    _doOperation("clear_hot", record, None, debug, limit, None, dry)
+    _doOperation("clear_hot", record, None, limit, None, dry)
 
 
 @cold.command()
 @with_appcontext
-@option_debug
-def check_transfers(debug):
+def process_transfers():
     """Check the status of the transfers."""
-    m = ColdStorageManager(current_app)
-    return m.check_current_transfers()
+    return TransferService.process_transfers()
 
 
 @cold.command()
 @with_appcontext
-@option_debug
-def check_requests(debug):
+def process_requests():
     """Check the status of the requests."""
-    m = ColdStorageManager(current_app)
-    return m.check_requests()
+
+    return RequestService.process_requests()