Add Completed Transfer Table

alembic/versions/84785333a677_add_completed_transfers_table.py

@@ -0,0 +1,49 @@
+# 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),
+        # --- Added missing columns ---
+        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_mbps", 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
 
+# importing new libraries for the complete transfers
+from datetime import datetime
+from typing import Optional
+from globus_sdk.authorizers import GlobusAuthorizer
+
 
 class GlobusAsyncTransferManager(CoreAsyncTransferManager):
     """
@@ -23,6 +28,10 @@
     destination_endpoint: str
     # The Globus endpoint UUID for the destination, entered in the configuration.
 
+    # I am adding this to allow one time authorization
+    authorizer: Optional[GlobusAuthorizer] = None
+    # ---
+
     native_app: bool = False
     # Whether to use a Native App (true) or a Confidential App (false, default)
     # for authorizing the client.
@@ -31,6 +40,9 @@
     transfer_complete: bool = False
     task_id: str = ""
 
+    #
+    model_config = {"arbitrary_types_allowed": True}
+
     @staticmethod
     def _subtract_local_root(path: Path, settings: "ServerSettings") -> Path:
         """
@@ -379,3 +391,68 @@
         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 an authorizer isn't passed in, use the original method.
+        # This allows our test to "inject" a working one.
+        # authorizer = self.authorize(settings=settings)
+
+        if self.authorizer is None:
+            print("--> No authorizer provided, attempting internal authorization...")
+            self.authorizer = self.authorize(settings=settings)
+
+        if not self.authorizer:
+            print("Authorization failed.")
+            return None
+
+        print("Authorization successful.")
+
+        transfer_client = globus_sdk.TransferClient(authorizer=self.authorizer)
+
+        try:
+            print(f"--> Fetching task details for ID: {self.task_id}")
+            task_doc = transfer_client.get_task(self.task_id)
+            print(f"Task data fetched. Status is: {task_doc['status']}")
+        except globus_sdk.TransferAPIError as e:
+            print(f"Globus API Error when fetching task: {e}")
+            return None
+
+        if task_doc["status"] != "SUCCEEDED":
+            print(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()
+
+        bandwidth_bps = bytes_transferred / duration_seconds
+
+        if duration_seconds > 0:
+            bandwidth_mbps = (bandwidth_bps * 8) / 1_000_000
+        else:
+            bandwidth_mbps = 0
+
+        # creating the report dict
+        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": round(bandwidth_mbps, 2),
+            }
+        except:
+            print("there is an error!")
+
+        return transfer_report

hera_librarian/async_transfers/local.py

@@ -1,3 +1,3 @@
 """
 The local async transfer manager.
 """
@@ -12,14 +12,38 @@
 
 from .core import CoreAsyncTransferManager
 
+# importing other libraries
+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
 
+    # Adding the transfer attributes
+    start_time_transfer: Optional[float] = None
+    bytes_transfer: Optional[int] = None
+
     def batch_transfer(self, paths: list[tuple[Path]], settings: "ServerSettings"):
+        # Begin initialization of time and total bytes
+        self.start_time_transfer = time.time()
+        total_bytes = 0
+
+        for local_path, _ in paths:
+            if local_path.is_dir():
+                total_bytes += sum(
+                    f.stat().st_size for f in local_path.glob("**/*") if f.is_file()
+                )
+            else:
+                total_bytes += local_path.stat().st_size
+
+        self.bytes_transfer = total_bytes
+        # End of initializations
+
         copy_success = True
 
         for local_path, remote_path in paths:
@@ -126,3 +150,36 @@
 
     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 = time.time()
+        duration_seconds = end_time - self.start_time_transfer
+        total_bytes = self.bytes_transfer
+
+        if duration_seconds > 0:
+            bandwidth_mbps = (total_bytes * 8) / (duration_seconds * 1_000_000)
+        else:
+            bandwidth_mbps = -1
+
+        # 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": datetime.fromtimestamp(
+                self.start_time_transfer, tz=timezone.utc
+            ),
+            "end_time": datetime.fromtimestamp(end_time, tz=timezone.utc),
+            "duration_seconds": duration_seconds,
+            "bytes_transferred": total_bytes,
+            "effective_bandwidth_mbps": round(bandwidth_mbps, 2),
+        }
+
+        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):
+    """
+    Class's information will be added here!
+    """
+
+    __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_mbps = db.Column(db.Integer, nullable=False)