Add Completed Transfer Table

alembic/versions/84785333a677_add_completed_transfers_table.py

@@ -0,0 +1,47 @@
+# Copyright 2017 the HERA Collaboration
+# Licensed under the 2-clause BSD License.
+
+"""Add completed_transfers table
+
+Revision ID: 84785333a677
+Revises: 38a604ac628b
+Create Date: 2025-07-25 00:51:04.843683
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "84785333a677"
+down_revision = "38a604ac628b"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "completed_transfers",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("task_id", sa.String(length=256), nullable=False),
+        sa.Column("source_endpoint_id", sa.String(length=256), nullable=False),
+        sa.Column("destination_endpoint_id", sa.String(length=256), nullable=False),
+        sa.Column("start_time", sa.DateTime(), nullable=False),
+        sa.Column("end_time", sa.DateTime(), nullable=False),
+        sa.Column("duration_seconds", sa.Float(), nullable=False),
+        sa.Column("bytes_transferred", sa.BigInteger(), nullable=False),
+        sa.Column("effective_bandwidth_bps", sa.Float(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["id"],
+            ["send_queue.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("task_id"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("completed_transfers")
+    # ### end Alembic commands ###

hera_librarian/async_transfers/globus.py

@@ -1,3 +1,3 @@
 """
 A transfer manager for Globus transfers.
 """
@@ -12,6 +12,12 @@
 
 from .core import CoreAsyncTransferManager
 
+# importing new libraries for the complete transfers
+from datetime import datetime
+from typing import Optional
+from globus_sdk.authorizers import GlobusAuthorizer
+from loguru import logger
+
 
 class GlobusAsyncTransferManager(CoreAsyncTransferManager):
     """
@@ -379,3 +385,63 @@
         except globus_sdk.TransferAPIError as e:
             return False
         return True
+
+    def complete_transfer(self, settings: "ServerSettings") -> dict | None:
+        """
+        Gathers details about a completed transfer from Globus and
+        returns them in a dictionary.
+        """
+
+        if self.authorizer is None:
+            logger.debug("--> No authorizer provided, attempting internal authorization...")
+            self.authorizer = self.authorize(settings=settings)
+
+        if not self.authorizer:
+            logger.error("Authorization failed.")
+            return None
+
+        logger.debug("Authorization successful.")
+
+        transfer_client = globus_sdk.TransferClient(authorizer=self.authorizer)
+
+        try:
+            logger.debug(f"--> Fetching task details for ID: {self.task_id}")
+            task_doc = transfer_client.get_task(self.task_id)
+            logger.debug(f"Task data fetched. Status is: {task_doc['status']}")
+        except globus_sdk.TransferAPIError as e:
+            logger.error(f"Globus API Error when fetching task: {e}")
+            return None
+
+        if task_doc["status"] != "SUCCEEDED":
+            logger.warning(f"Task status is not SUCCEEDED.")
+            return None
+
+        # Calculate duration and bandwidth
+        bytes_transferred = task_doc["bytes_transferred"]
+        start_time = datetime.fromisoformat(task_doc["request_time"])
+        end_time = datetime.fromisoformat(task_doc["completion_time"])
+
+        duration = end_time - start_time
+        duration_seconds = duration.total_seconds()
+
+        if duration_seconds > 0:
+            bandwidth_bps = bytes_transferred / duration_seconds
+        else:
+            # Handle the zero-duration case
+            bandwidth_mbps = 0.0
+
+        try:
+            transfer_report = {
+                "task_id": task_doc["task_id"],
+                "source_endpoint_id": task_doc["source_endpoint_id"],
+                "destination_endpoint_id": task_doc["destination_endpoint_id"],
+                "start_time": start_time,
+                "end_time": end_time,
+                "duration_seconds": duration_seconds,
+                "bytes_transferred": bytes_transferred,
+                "effective_bandwidth_mbps": bandwidth_mbps,
+            }
+        except (KeyError, ValueError) as e:
+            logger.error(f"Missing key or malformed value: {e} related to task {self.task_id}")
+
+        return transfer_report

hera_librarian/async_transfers/local.py

@@ -1,3 +1,3 @@
 """
 The local async transfer manager.
 """
@@ -10,16 +10,33 @@
 
 from hera_librarian.transfer import TransferStatus
 
+from hera_librarian.utils import get_size_from_path
+
 from .core import CoreAsyncTransferManager
 
+import time
+from datetime import datetime, timezone
+from typing import Optional
+
 
 class LocalAsyncTransferManager(CoreAsyncTransferManager):
     hostnames: list[str]
 
     transfer_attempted: bool = False
     transfer_complete: bool = False
 
+    # Attributes to store performance metrics for the completion report
+    start_time_transfer: Optional[datetime] = None
+    bytes_transfer: Optional[int] = None
+
     def batch_transfer(self, paths: list[tuple[Path]], settings: "ServerSettings"):
+        # Initialization of time and total bytes
+        self.start_time_transfer = datetime.now(timezone.utc)
+        # Use the existing utility function to calculate the total size
+        total_bytes = sum(get_size_from_path(local_path) for local_path, _ in paths)
+
+        self.bytes_transfer = total_bytes
+
         copy_success = True
 
         for local_path, remote_path in paths:
@@ -126,3 +143,34 @@
 
     def fail_transfer(self, settings: "ServerSettings") -> bool:
         return True
+
+    def complete_transfer(self) -> dict | None:
+        """ """
+        # 1. Check if the transfer metrics were recorded
+        if self.start_time_transfer is None or self.bytes_transfer is None:
+            print("Error: Transfer metrics were not recorded.")
+            return None
+
+        # 2. Calculate performance metrics from the recorded data
+        end_time = datetime.now(timezone.utc)
+        duration_seconds = (end_time - self.start_time_transfer).total_seconds()
+        total_bytes = self.bytes_transfer
+
+        if duration_seconds > 0:
+            bandwidth_bps = total_bytes / duration_seconds
+        else:
+            bandwidth_bps = 0
+
+        # 3. Create and return the report dictionary
+        transfer_report = {
+            "task_id": f"local_{end_time}",
+            "source_endpoint_id": gethostname(),
+            "destination_endpoint_id": gethostname(),
+            "start_time": self.start_time_transfer,
+            "end_time": end_time,
+            "duration_seconds": duration_seconds,
+            "bytes_transferred": total_bytes,
+            "effective_bandwidth_bps": bandwidth_bps,
+        }
+
+        return transfer_report

librarian_background/queues.py

@@ -1,3 +1,3 @@
 """
 Consumers for the background task queues. Note that these must run in a separate
 thread than the main background tasks to avoid blocking during long-running
@@ -26,6 +26,8 @@
 
 from .task import Task
 
+from librarian_server.orm.sendqueue import CompletedTransfer
+
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
 
@@ -215,6 +217,22 @@
                 )
                 continue
 
+            # If the transfer is complete, generate and save the performance report.
+            # This works for both local and globus transfers.
+            if current_status == TransferStatus.COMPLETED:
+                report_dict = queue_item.async_transfer_manager.complete_transfer()
+
+                if report_dict:
+                    logger.info(
+                        "Creating completion record for queue item {q.id}", q=queue_item
+                    )
+                    # Add the queue item's ID to the report to create the one-to-one link
+                    report_dict["id"] = queue_item.id
+
+                    # Create the record using the CompletedTransfer model
+                    completed_record = CompletedTransfer(**report_dict)
+                    session.add(completed_record)
+
             # If we got down here, we can mark the transfer as consumed.
             logger.info("Marking {q.id} as completed", q=queue_item)
             queue_item.completed = True

librarian_server/orm/__init__.py

@@ -1,3 +1,3 @@
 """
 ORM for database storage.
 """
@@ -6,7 +6,7 @@
 from .file import CorruptFile, File
 from .instance import Instance, RemoteInstance
 from .librarian import Librarian
-from .sendqueue import SendQueue
+from .sendqueue import SendQueue, CompletedTransfer
 from .storemetadata import StoreMetadata
 from .transfer import CloneTransfer, IncomingTransfer, OutgoingTransfer, TransferStatus
 from .user import User

librarian_server/orm/sendqueue.py

@@ -73,6 +73,13 @@ class SendQueue(db.Base):
     failed: bool = db.Column(db.Boolean, default=False)
     "Whether this queue item failed, and that is the reason for completed status."
 
+    completed_record = db.relationship(
+        "CompletedTransfer",
+        uselist=False,
+        back_populates="send_queue",
+        cascade="all, delete-orphan",
+    )
+
     @classmethod
     def new_item(
         cls,
@@ -204,3 +211,31 @@ def update_transfer_status(
         session.commit()
 
         return response
+
+
+class CompletedTransfer(db.Base):
+    """
+    Stores a record of a successfuly completed transfer.
+    """
+
+    __tablename__ = "completed_transfers"
+
+    id = db.Column(db.Integer, db.ForeignKey("send_queue.id"), primary_key=True)
+
+    send_queue = db.relationship("SendQueue", back_populates="completed_record")
+
+    task_id = db.Column(db.String(256), nullable=False, unique=True)
+
+    source_endpoint_id = db.Column(db.String(256), nullable=False)
+
+    destination_endpoint_id = db.Column(db.String(256), nullable=False)
+
+    start_time = db.Column(db.DateTime, nullable=False)
+
+    end_time = db.Column(db.DateTime, nullable=False)
+
+    duration_seconds = db.Column(db.Integer, nullable=False)
+
+    bytes_transferred = db.Column(db.BigInteger, nullable=False)
+
+    effective_bandwidth_bps = db.Column(db.Integer, nullable=False)

tests/integration_test/test_send_queue.py

@@ -31,6 +31,9 @@ def transfer_status(self, *args, **kwargs):
     def fail_transfer(self, settings):
         return True
 
+    def complete_transfer(self):
+        return None
+
 
 def test_create_simple_queue_item_and_send(
     test_server, test_orm, mocked_admin_client, server
@@ -72,6 +75,7 @@ def test_create_simple_queue_item_and_send(
     from librarian_background.queues import check_on_consumed, consume_queue_item
 
     consume_queue_item(session_maker=get_session)
+
     check_on_consumed(
         session_maker=get_session,
         timeout_after=datetime.now(timezone.utc) + timedelta(days=7),
@@ -717,3 +721,100 @@ def test_create_send_queue_item_no_availability_of_transfer_manager(
         assert test == False
 
     return
+
+
+def test_local_transfer_creates_completion_record(
+    test_server_with_valid_file, test_orm, mocked_admin_client, server, tmp_path
+):
+    """
+    Tests that a successful local transfer processed by the queue
+    creates a corresponding record in the completed_transfers table.
+    """
+    # The fixtures provide us with a running server, a database session,
+    # and a pre-existing file to transfer.
+    SendQueue = test_orm.SendQueue
+    File = test_orm.File
+    OutgoingTransfer = test_orm.OutgoingTransfer
+    CompletedTransfer = test_orm.CompletedTransfer
+    get_session = test_server_with_valid_file[1]
+
+    # Adding this block to register the destination librarian
+    mocked_admin_client.add_librarian(
+        name="local_test_server",
+        url="http://localhost",
+        authenticator="admin:password",
+        port=server.id,
+    )
+
+    with get_session() as session:
+        # Get the file provided by the fixture
+        file = session.get(File, "example_file.txt")
+
+        # Create an OutgoingTransfer record for this file
+        transfer = OutgoingTransfer.new_transfer(
+            destination="local_test_server",
+            instance=file.instances[0],
+            file=file,
+        )
+
+        # Define the source and destination paths for the local copy
+        transfer.source_path = str(file.instances[0].path)
+        transfer.dest_path = str(tmp_path / file.name)
+
+        # Create the SendQueue item for this transfer
+        # using the real LocalAsyncTransferManager.
+        queue_item = SendQueue.new_item(
+            priority=100,
+            destination="local_test_server",
+            transfers=[transfer],
+            async_transfer_manager=LocalAsyncTransferManager(hostnames=[gethostname()]),
+        )
+
+        session.add_all([transfer, queue_item])
+        session.commit()
+        # Keep the ID for later verification
+        queue_item_id = queue_item.id
+        transfer_id = transfer.id
+        source_path_for_check = transfer.source_path
+
+    # These are the core functions that the librarian's background
+    # process would run.
+    from librarian_background.queues import check_on_consumed, consume_queue_item
+
+    print("Consuming queue item to start transfer...")
+    consume_queue_item(session_maker=get_session)
+
+    print("Checking on consumed item to finalize and record completion...")
+    check_on_consumed(
+        session_maker=get_session,
+        timeout_after=datetime.now(timezone.utc) + timedelta(days=7),
+    )
+
+    with get_session() as session:
+        # First, confirm the original SendQueue job was marked as complete
+        completed_queue_item = session.get(SendQueue, queue_item_id)
+        assert completed_queue_item.consumed
+        assert completed_queue_item.completed
+
+        # Re-fetch the transfer object to ensure it's attached to this session
+        transfer = session.query(OutgoingTransfer).filter_by(id=transfer.id).one()
+
+        # Now, check if our new CompletedTransfer record was created.
+        # This is the main goal of our test.
+        completion_record = (
+            session.query(CompletedTransfer).filter_by(id=queue_item_id).one_or_none()
+        )
+
+        # Assert that the record exists and has the correct data
+        assert (
+            completion_record is not None
+        ), "A CompletedTransfer record should have been created."
+        assert (
+            completion_record.bytes_transferred
+            == Path(transfer.source_path).stat().st_size
+        )
+        assert completion_record.effective_bandwidth_bps > 0
+
+        print(
+            f"\n Success! CompletedTransfer record created with ID: {completion_record.id}"
+        )