medn

Author: psaiz

cernopendata/cold_storage/api.py

@@ -91,7 +91,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 +110,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
@@ -261,8 +262,8 @@ def check_requests(self, manager):
     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
+            active_transfers_count = TransferMetadata.query.filter(
+                TransferMetadata.finished.is_(None), TransferMetadata.action == action
             ).count()
             threshold = Request.get_active_transfers_threshold(action)
 
@@ -272,7 +273,7 @@ def _check_submitted(manager):
             submitted = 0
             limit = threshold - active_transfers_count
             if limit > 0:
-                transfers = Request.query.filter_by(
+                transfers = RequestMetadata.query.filter_by(
                     status="submitted", action=action
                 ).all()
 
@@ -291,6 +292,7 @@ def _check_submitted(manager):
                         transfer.num_files += len(info)
                         transfer.size += sum(item.size for item in info)
                     transfer.started_at = datetime.utcnow()
+                    logger.info(f"THE LIMIT WAS {limit}, AND WE SUBMITTED {submitted}")
                     if limit == submitted:
                         logger.info(
                             f"Reached the threshold of {threshold} transfers. There might be more in this record"
@@ -312,7 +314,9 @@ 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()
+            requests = RequestMetadata.query.filter_by(
+                status="started", action=action
+            ).all()
             logger.debug(f"Checking the {len(requests)} {action} requests")
             completed = 0
             for request in requests:
@@ -352,22 +356,20 @@ def get_requests(
         """Get the summary of the requests."""
         if summary:
             query = db.session.query(
-                Request.status,
-                Request.action,
+                RequestMetadata.status,
+                RequestMetadata.action,
                 func.count().label("count"),
-                func.sum(Request.num_files).label("files"),
-                func.sum(Request.size).label("size"),
+                func.sum(RequestMetadata.num_files).label("files"),
+                func.sum(RequestMetadata.size).label("size"),
             )
         else:
-            query = Request.query
+            query = RequestMetadata.query
 
         if status:
-            status_list = status.split(",")
-            query = query.filter(Request.status.in_(status_list))
+            query = query.filter(RequestMetadata.status.in_(status))
 
         if action:
-            action_list = action.split(",")
-            query = query.filter(Request.action.in_(action_list))
+            query = query.filter(RequestMetadata.action.in_(action))
 
         if record:
             try:
@@ -376,7 +378,9 @@ def get_requests(
             except Exception:
                 query = query.filter(False)
         if summary:
-            result = query.group_by(Request.status, Request.action).all()
+            result = query.group_by(
+                RequestMetadata.status, RequestMetadata.action
+            ).all()
 
         if sort:
             column = getattr(Request, sort, None)
@@ -387,7 +391,7 @@ def get_requests(
                     query = query.order_by(column.asc())
 
         if page:
-            result = query.order_by(Request.created_at.desc()).paginate(
+            result = query.order_by(RequestMetadata.created_at.desc()).paginate(
                 page=page, per_page=per_page, error_out=False
             )
 

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

@@ -47,7 +47,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 +79,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 +111,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 +130,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 +152,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 +185,22 @@ 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 check_transfers():
     """Check the status of the transfers."""
     m = ColdStorageManager(current_app)
     return m.check_current_transfers()
 
 
 @cold.command()
 @with_appcontext
-@option_debug
-def check_requests(debug):
+def check_requests():
     """Check the status of the requests."""
     m = ColdStorageManager(current_app)
     return m.check_requests()

cernopendata/cold_storage/manager.py

@@ -125,10 +125,12 @@ def _move_record(
         record = self._catalog.get_record(record_uuid)
         if not record:
             return []
-        for file in self._catalog.get_files_from_record(record, limit):
+        for file in self._catalog.get_files_from_record(record):
             transfers += self._move_record_file(
                 record.id, file, action, move_function, register, force, dry
             )
+            if len(transfers) >= limit:
+                logger.info("Reached the limit. Going back")
         logger.info(f"{len(transfers)} transfers have been issued")
         return transfers
 

cernopendata/cold_storage/schemas.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of CERN Open Data Portal.
+# Copyright (C) 2017-2025 CERN.
+#
+# CERN Open Data Portal is free software; you can redistribute it
+# and/or modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# CERN Open Data Portal is distributed in the hope that it will be
+# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with CERN Open Data Portal; if not, write to the
+# Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
+# MA 02111-1307, USA.
+#
+# In applying this license, CERN does not
+# waive the privileges and immunities granted to it by virtue of its status
+# as an Intergovernmental Organization or submit itself to any jurisdiction.
+
+"""Schemas used for the transfers."""
+
+from invenio_pidstore.models import PersistentIdentifier
+from marshmallow import Schema, fields, validate
+
+
+class TransferRequestQuerySchema(Schema):
+    """Arguments for the transfer search."""
+
+    page = fields.Int(missing=1)
+    per_page = fields.Int(missing=10)
+    status = fields.List(fields.Str())
+    action = fields.List(fields.Str())
+    record = fields.Str()
+    sort = fields.Str()
+    direction = fields.Str(validate=validate.OneOf(["asc", "desc"]), missing="asc")
+
+
+class TransferRequestSchema(Schema):
+    """Schema of a transfer."""
+
+    id = fields.Str()
+    recid = fields.Method("get_recid")
+    action = fields.Str()
+    num_files = fields.Int()
+    size = fields.Int()
+    status = fields.Str()
+    created_at = fields.DateTime()
+    started_at = fields.DateTime(allow_none=True)
+    completed_at = fields.DateTime(allow_none=True)
+
+    def get_recid(self, obj):
+        """Convert the uuid into the recid."""
+        pid = PersistentIdentifier.query.filter_by(object_uuid=obj.record_id).first()
+        return pid.pid_value if pid else None

cernopendata/config.py

@@ -68,32 +68,10 @@
 # Logging - Set up Sentry for Invenio-Logging
 SENTRY_DSN = os.environ.get("SENTRY_DSN", None)
 
-LOGGING = {
-    "version": 1,
-    "disable_existing_loggers": False,
-    "formatters": {
-        "default": {
-            "format": "[%(asctime)s] %(levelname)s in %(module)s: %(message)s",
-        },
-    },
-    "handlers": {
-        "console": {
-            "class": "logging.StreamHandler",
-            "formatter": "default",
-        },
-    },
-    "root": {
-        "level": "INFO",  # <- Set global level to INFO
-        "handlers": ["console"],
-    },
-    "loggers": {
-        "invenio": {
-            "level": "INFO",  # <- Set Invenio-specific logging level
-            "handlers": ["console"],
-            "propagate": False,
-        },
-    },
-}
+COLD_ACTIVE_ARCHIVING_TRANSFERS_THRESHOLD = 1000
+# Maximum number of transfers that should be active at a given moment
+COLD_ACTIVE_STAGING_TRANSFERS_THRESHOLD = 50
+# Maximum number of transfers that should be active at a given moment
 
 LOGGING_SENTRY_CELERY = os.environ.get("LOGGING_SENTRY_CELERY", False)