First version of the cold storage interface. New actions to request files, see the requests and subscribe

Author: psaiz

Dockerfile

@@ -29,6 +29,11 @@ FROM --platform=$BUILDPLATFORM registry.cern.ch/inveniosoftware/almalinux:1
 # Use XRootD 5.8.1
 ENV XROOTD_VERSION=5.8.1
 
+# Install the CERN CA
+COPY docker/carepo.repo /etc/yum.repos.d/
+
+RUN yum install -y ca_CERN-Root-2 && yum clean -y all
+
 # Install CERN Open Data Portal web node pre-requisites
 # hadolint ignore=DL3033
 RUN yum install -y \
@@ -42,8 +47,9 @@ RUN yum install -y \
     yum groupinstall -y "Development Tools" && \
     yum clean -y all
 
+# hadolint ignore=DL3033
 RUN echo "Will install xrootd version: $XROOTD_VERSION (latest if empty)" && \
-    yum install -y xrootd-"$XROOTD_VERSION" python3-xrootd-"$XROOTD_VERSION" && \
+    yum install -y xrootd-"$XROOTD_VERSION" python3-xrootd-"$XROOTD_VERSION" swig python3-gfal2-util gfal2-plugin-http python3-gfal2 && \
     yum clean -y all
 
 RUN pip uninstall pipenv -y && pip install --no-cache-dir --upgrade pip==24.3.1 setuptools==70.0.0 wheel==0.45.1 && \
@@ -72,13 +78,16 @@ ENV PATH=$PATH:${INVENIO_INSTANCE_PATH}/python/bin
 
 # Add CERN Open Data Portal sources to `code` and work there
 WORKDIR ${CODE_DIR}
+COPY . ${CODE_DIR}
+USER root
+RUN chown -R "${INVENIO_USER_ID}":root "${CODE_DIR}"
+USER ${INVENIO_USER_ID}
 
 # Debug off by default
 ARG DEBUG=""
 ENV DEBUG=${DEBUG:-""}
 
 # Install CERN Open Data Portal sources
-COPY . ${CODE_DIR}
 # hadolint ignore=DL3013,SC2086
 RUN git config --global url.https://github.com/.insteadOf git://github.com/ && if [ "$DEBUG" ]; then FLAGS="-e"; fi && \
     pip install --no-cache-dir --user ${FLAGS} ".[all]" && pip check

MANIFEST.in

@@ -45,3 +45,4 @@ recursive-include sentry *.json
 recursive-include sentry *.py
 recursive-include sentry *.sh
 recursive-include sentry *.yml
+recursive-include docker *.repo

cernopendata/api.py

@@ -45,7 +45,7 @@ def __getitem__(self, key):
         """Get a specific file."""
         obj = FileIndexMetadata.get(self.record, key)
         if obj:
-            return self.file_cls(obj, self.filesmap.get(obj.key, {}))
+            return self.file_cls(obj, self.file_indices.get(obj.key, {}))
         raise KeyError(key)
 
     def flush(self):
@@ -98,6 +98,17 @@ def check_availability(self):
         else:
             self["availability"] = "sample files"
 
+    def flush_indices(self):
+        """Updates the _file_indices information based on what exists on the database."""
+        print("Updating the record with file indices")
+        self["_file_indices"] = []
+        # First, let's get all the file indices that this record has
+        for elem in BucketTag.query.filter_by(value=str(self.id), key="record").all():
+            self["_file_indices"].append(
+                FileIndexMetadata.get(None, str(elem.bucket)).dumps()
+            )
+        self.check_availability()
+
 
 class FileIndexMetadata:
     """Class for the FileIndexMetadata."""
@@ -109,6 +120,7 @@ def __init__(self):
         self._size = 0
         self._files = []
         self._description = ""
+        self._bucket = ""
 
     def __repr__(self):
         """Representation of the object."""
@@ -131,6 +143,7 @@ def create(cls, record, file_object, description=""):
         rb._description = description
         BucketTag.create(rb._bucket, "index_name", index_file_name)
         BucketTag.create(rb._bucket, "record", record.model.id)
+        BucketTag.create(rb._bucket, "description", description)
         print(f"The file index contains {len(index_content)} entries.")
         for entry in index_content:
             entry_file = FileInstance.create()
@@ -162,9 +175,13 @@ def get(cls, record_id, bucket_id):
             .one()
             .value
         )
-        bucket = Bucket.get(bucket_id)
-        for o in ObjectVersion.get_by_bucket(bucket).all():
-            f = FileObject(o, {})
+        tag = BucketTag.query.filter_by(
+            bucket_id=str(bucket_id), key="description"
+        ).first()
+        obj._description = tag.value if tag else obj._index_file_name
+        obj._bucket = Bucket.get(bucket_id)
+        for o in ObjectVersion.get_by_bucket(obj._bucket).all():
+            f = MultiURIFileObject(o, {})
             # Let's put also the uri
             f["uri"] = FileInstance.get(str(o.file_id)).uri
             f["filename"] = f["uri"].split("/")[-1]
@@ -196,6 +213,7 @@ def dumps(self):
             "size": self._size,
             "files": files,
             "description": self._description,
+            "bucket": str(self._bucket),
         }
 
     def flush(self):

cernopendata/cold_storage/__init__.py

@@ -0,0 +1,24 @@
+# -*- 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.
+"""Cold Storage."""

cernopendata/cold_storage/api.py

@@ -0,0 +1,73 @@
+# -*- 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.
+
+"""Cold Storage API."""
+import json
+from collections import OrderedDict
+
+from invenio_files_rest.models import (
+    Bucket,
+    BucketTag,
+    FileInstance,
+    ObjectVersion,
+    ObjectVersionTag,
+)
+from invenio_records_files.api import FileObject, FilesIterator, Record
+from invenio_records_files.models import RecordsBuckets
+
+
+class FileObjectCold(FileObject):
+    """Overwrite the fileobject to get multiple URI."""
+
+    @classmethod
+    def create(self, bucket, filename, file_id):
+        """Create a cold file object."""
+        return ObjectVersion.create(bucket, filename, file_id)
+
+    def dumps(self):
+        """This one has the information about the cold URI stored in a ObjectVersionTag."""
+        info = super(FileObjectCold, self).dumps()
+        info["tags"] = {}
+        for tagName in ("uri_cold", "hot_deleted"):
+            tag = ObjectVersionTag.get(str(self.obj.version_id), tagName)
+            if tag:
+                info["tags"][tagName] = tag.value
+        if "availability" in self.data:
+            del self.data["availability"]
+        info["availability"] = self.availability
+        if "uri" not in info:
+            file = FileInstance.get(str(self.obj.file_id))
+            info["uri"] = file.uri
+        return info
+
+    @property
+    def availability(self):
+        """Describe the QoS of the file."""
+        if "availability" not in self.data:
+            avl = "ready"
+            for t in self.obj.tags:
+                if t.key == "hot_deleted":
+                    avl = "needs request"
+            self.data["availability"] = avl
+        return self.data["availability"]

cernopendata/cold_storage/catalog.py

@@ -0,0 +1,139 @@
+# -*- 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.
+
+"""Cold Storage Catalog."""
+
+from datetime import datetime
+
+from invenio_db import db
+from invenio_files_rest.models import FileInstance, ObjectVersion, ObjectVersionTag
+from invenio_indexer.api import RecordIndexer
+from invenio_pidstore.models import PersistentIdentifier
+
+from cernopendata.api import RecordFilesWithIndex
+
+
+class Catalog:
+    """Class to interact with the repository."""
+
+    def __init__(self, debug=False):
+        """Initialize the catalog."""
+        self._indexer = RecordIndexer()
+        self._debug = debug
+        self._reindex_queue = []
+
+    def get_record(self, record_uuid):
+        """First, lets get the record."""
+        try:
+            record = RecordFilesWithIndex.get_record(record_uuid)
+        except Exception as e:
+            print(f"Couldn't find a record  with the id '{record_uuid}': {e}")
+            return
+        return record
+
+    def get_files_from_record(self, record, limit=None):
+        """Getting the files from a record."""
+        files = []
+        if self._debug:
+            print("DEBUG: The catalog got the record:", record.to_dict())
+        if record:
+            if "_files" in record:
+                files += record["_files"]
+            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]
+        if self._debug:
+            print("DEBUG: And the list of files are:", files)
+        return files
+
+    def clear_hot(self, record, file_id):
+        """Marking the hot copy as deleted."""
+
+        def _clear_hot_function(version_id):
+            """Create a tag for the file identifying that the copy is not available."""
+            ObjectVersionTag.create(version_id, "hot_deleted", str(datetime.now()))
+
+        return self._update_file_and_reindex(record.id, file_id, _clear_hot_function)
+
+    def _update_file_and_reindex(self, record_uuid, file_id, update_function):
+        """Function to update the repository."""
+        f = FileInstance.get(file_id)
+        if not f:
+            print(f"Can't find that file :( {file_id}")
+            return False
+        objectVersion = ObjectVersion.query.filter_by(file_id=f.id).one_or_none()
+        if not objectVersion:
+            print(f"Can't find the object associated to that file :( {file_id}")
+            return False
+        update_function(objectVersion.version_id)
+        db.session.commit()
+        if record_uuid not in self._reindex_queue:
+            print(f"Record {record_uuid} will be reindexed")
+            self._reindex_queue += [record_uuid]
+        return True
+
+    def reindex_entries(self):
+        """Reindexes all the entries that have been modified."""
+        while len(self._reindex_queue) > 0:
+            record_uuid = self._reindex_queue.pop(0)
+            print(f"Ready to reindex {record_uuid}")
+            record = RecordFilesWithIndex.get_record(record_uuid)
+            if not record:
+                print(f"Couldn't find that record '{record_uuid}'")
+                continue
+            print("Got the object from the database")
+            record.files.flush()
+            record.flush_indices()
+            record.commit()
+            db.session.commit()
+            try:
+                self._indexer.index(record)
+            except Exception as e:
+                print(f"Error during the reindex {e}")
+                try:
+                    record.commit()
+                    self._indexer.index(record)
+                    print("The second time worked!")
+                except Exception as e:
+                    print(f"Doing it again did not help :( {e}")
+
+    def add_copy(self, record_uuid, file_id, new_qos, new_filename):
+        """Adds a copy to a particular file. It reindexes the record."""
+
+        def _add_copy_function(version_id):
+            """Function to add a file tag with a new uri for the file."""
+            if new_qos == "cold":
+                ObjectVersionTag.create_or_update(version_id, "uri_cold", new_filename)
+            elif new_qos == "hot":
+                ObjectVersionTag.delete(version_id, "hot_deleted")
+
+        return self._update_file_and_reindex(record_uuid, file_id, _add_copy_function)