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,11 @@
 
 from .core import CoreAsyncTransferManager
 
+from datetime import datetime
+from typing import Optional
+from loguru import logger
+from hera_librarian.models.transfer import CompletedTransferCore
+
 
 class GlobusAsyncTransferManager(CoreAsyncTransferManager):
     """
@@ -379,3 +384,62 @@
         except globus_sdk.TransferAPIError as e:
             return False
         return True
+
+    def gather_transfer_details(
+        self, settings: "ServerSettings"
+    ) -> CompletedTransferCore | None:
+        """
+        Gathers details about a completed transfer from Globus and
+        returns them in a Pydantic object.
+        """
+
+        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
+
+        start_time = datetime.fromisoformat(task_doc["request_time"])
+        end_time = datetime.fromisoformat(task_doc["completion_time"])
+        bytes_transferred = task_doc["bytes_transferred"]
+        bandwidth_bps = task_doc["effective_bytes_per_second"]
+
+        duration = end_time - start_time
+        duration_seconds = duration.total_seconds()
+
+        try:
+            transfer_record = CompletedTransferCore(
+                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_bps,
+            )
+            return transfer_record
+        except (KeyError, ValueError) as e:
+            logger.error(
+                f"Missing key or malformed value: {e} related to task {self.task_id}"
+            )

hera_librarian/async_transfers/local.py

@@ -1,3 +1,3 @@
 """
 The local async transfer manager.
 """
@@ -10,16 +10,34 @@
 
 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
+from hera_librarian.models.transfer import CompletedTransferCore
+
 
 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 +144,40 @@
 
     def fail_transfer(self, settings: "ServerSettings") -> bool:
         return True
+
+    def gather_transfer_details(self) -> CompletedTransferCore | None:
+        """
+        Gathers details about a locall completed transfer and
+        returns them in a CompletedTransferCore Pydantic Object
+        """
+        # Check if the transfer metrics were recorded
+        if self.start_time_transfer is None or self.bytes_transfer is None:
+            logger.error("Error: Transfer metrics were not recorded")
+            return None
+
+        # 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:
+            duration_seconds = 1
+
+        bandwidth_bps = total_bytes / duration_seconds
+
+        try:
+            transfer_record = CompletedTransferCore(
+                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_record
+        except (KeyError, ValueError) as e:
+            logger.error(
+                f"Failed to create transfer report object due to a data validation error: {e}"
+            )

hera_librarian/models/transfer.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel
+from datetime import datetime
+
+class CompletedTransferCore(BaseModel):
+    """
+    A Pydantic model representing the data for a completed transfer.
+    """
+
+    task_id: str
+    source_endpoint_id: str
+    destination_endpoint_id: str
+    start_time: datetime
+    end_time: datetime
+    duration_seconds: float 
+    bytes_transferred: int 
+    effective_bandwidth_bps: float

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,9 @@
 
 from .task import Task
 
+from librarian_server.orm.completed_transfer import CompletedTransfer
+
+
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
 
@@ -215,6 +218,23 @@
                 )
                 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_core_model = (
+                    queue_item.async_transfer_manager.gather_transfer_details()
+                )
+
+                if report_core_model:
+                    logger.info(
+                        "Creating completion record for queue item {q.id}", q=queue_item
+                    )
+                    # use the classmethod to create the ORM object.
+                    completed_record = CompletedTransfer.from_core(
+                        core=report_core_model, queue_id=queue_item.id
+                    )
+                    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.
 """
@@ -7,6 +7,7 @@
 from .instance import Instance, RemoteInstance
 from .librarian import Librarian
 from .sendqueue import SendQueue
+from .completed_transfer import CompletedTransfer
 from .storemetadata import StoreMetadata
 from .transfer import CloneTransfer, IncomingTransfer, OutgoingTransfer, TransferStatus
 from .user import User

librarian_server/orm/completed_transfer.py

@@ -0,0 +1,48 @@
+from .. import database as db
+import datetime
+from hera_librarian.models.transfer import CompletedTransferCore
+
+class CompletedTransfer(db.Base):
+    """
+    The SQLAlchemy ORM model for a completed transfer.
+    """
+    __tablename__ = "completed_transfers"
+
+    id: int = db.Column(db.Integer, db.ForeignKey("send_queue.id"), primary_key=True)
+
+    send_queue = db.relationship("SendQueue", back_populates="completed_record")
+
+    task_id: str = db.Column(db.String(256), nullable=False, unique=True)
+
+    source_endpoint_id: str = db.Column(db.String(256), nullable=False)
+
+    destination_endpoint_id = db.Column(db.String(256), nullable=False)
+
+    start_time: datetime.datetime = db.Column(db.DateTime, nullable=False)
+
+    end_time: datetime.datetime = db.Column(db.DateTime, nullable=False)
+
+    duration_seconds: float = db.Column(db.Integer, nullable=False)
+
+    bytes_transferred: int = db.Column(db.BigInteger, nullable=False)
+
+    effective_bandwidth_bps: float = db.Column(db.Integer, nullable=False)
+
+    send_queue = db.relationship("SendQueue", back_populates="completed_record")
+
+    @classmethod
+    def from_core(cls, core: CompletedTransferCore, queue_id: int) -> "CompletedTransfer":
+        """
+        Creates a new CompletedTransfer ORM object from a core model.
+        """
+        return cls(
+            id=queue_id,
+            task_id=core.task_id,
+            source_endpoint_id=core.source_endpoint_id,
+            destination_endpoint_id=core.destination_endpoint_id,
+            start_time=core.start_time,
+            end_time=core.end_time,
+            duration_seconds=core.duration_seconds,
+            bytes_transferred=core.bytes_transferred,
+            effective_bandwidth_bps=core.effective_bandwidth_bps,
+        )

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,