import datetime
import os
import subprocess
import hashlib
import base64
import json
from typing import Tuple, Optional
import time
import tempfile
import coscine
from sqlalchemy.orm import Session
from ccat_ops_db import models
from .database import DatabaseConnection
from .setup_celery_app import app, make_celery_task
from .utils import (
get_s3_client,
get_redis_connection,
get_s3_key_for_package,
safe_join,
resolve_s3_method,
resolve_coscine_config,
deduplicate_by_id,
)
from .logging_utils import get_structured_logger
from .exceptions import OperationNotFoundError, ScheduleError
from .boundary import ensure_readable, diagnose_source_missing
from .metrics import HousekeepingMetrics
from .queue_discovery import route_task_by_location
from .operation_types import OperationType
from .task_state_manager import TaskStateManager
# Use only task loggers
logger = get_structured_logger(__name__)
redis_ = get_redis_connection()
# s3_client = boto3.client(
# "s3",
# endpoint_url=ccat_data_transfer_settings.s3_endpoint_url,
# aws_access_key_id=ccat_data_transfer_settings.s3_access_key_id,
# aws_secret_access_key=ccat_data_transfer_settings.s3_secret_access_key,
# region_name=ccat_data_transfer_settings.s3_region_name,
# config=boto3.session.Config(signature_version="s3v4"),
# )
[docs]
class LongTermArchiveTask(make_celery_task()):
"""Base class for long term archive tasks.
Drives the uniform ``ArchiveOperation`` (OperationKind.ARCHIVE): the archive
lifecycle -- status, ``retry_count``, consume/produce lineage -- lives on the
Operation row, not on the legacy ``LongTermArchiveTransfer`` (which remains
additively until #95). The task argument is the ArchiveOperation id.
"""
[docs]
def __init__(self):
super().__init__()
# Identity sourced from the enum (single source of truth), kept as the
# plain .value so TaskStateManager / circuit-breaker f-string key builds
# stay byte-identical -- a bare OperationKind member would render its NAME
# in an f-string under Python 3.12 and corrupt the Redis key.
self.operation_type = models.OperationKind.ARCHIVE.value
# No get_retry_count override: the archive operation carries the uniform
# Operation.retry_count, so it falls through to the base reader (#75/#155).
[docs]
def mark_in_progress(self, session, operation_id):
"""Set the ArchiveOperation IN_PROGRESS at task start (the base-task path).
This is the single seam through which the running signal is set; the
archive-local IN_PROGRESS set in the task body is removed (#155). It is
the honest running marker, NOT the duplicate guard -- correctness rests
on idempotent completion (ADR-0004): the COMPLETED early-return and the
existing-destination-copy no-op.
The seam runs (and commits) BEFORE the task body, so it must never
resurrect a settled operation: a redelivered/recovery-re-dispatched run
of an already-terminal op is left untouched, otherwise the COMPLETED ->
IN_PROGRESS flip would defeat the body's COMPLETED early-return and a
post-retention re-delivery could re-run the transfer (ADR-0004).
"""
archive_operation = session.get(models.ArchiveOperation, operation_id)
if archive_operation and archive_operation.status not in (
models.Status.COMPLETED,
models.Status.FAILED,
):
archive_operation.status = models.Status.IN_PROGRESS
[docs]
def reset_state_on_failure(self, session, archive_operation_id, exc):
"""Reset the archive operation state for retry."""
archive_operation = session.get(models.ArchiveOperation, archive_operation_id)
if archive_operation:
archive_operation.status = models.Status.PENDING
raw_data_package = _archive_raw_data_package(session, archive_operation)
if raw_data_package:
raw_data_package.state = models.PackageState.TRANSFERRING
archive_operation.failure_error_message = None
# error_context mirrors failure_error_message: NULLed on reset (#117);
# the durable trail is kept in OperationFailureEvent.
archive_operation.error_context = None
archive_operation.retry_count += 1
logger.info(
"reset archive operation for retry",
archive_operation_id=archive_operation_id,
retry_count=archive_operation.retry_count,
)
_add_archive_log(
session,
archive_operation,
f"Transfer failed, scheduling retry: {str(exc)}",
)
session.commit()
redis_.publish(
"transfer:overview",
json.dumps(
{
"type": "long_term_archive_transfer_reset",
"data": archive_operation_id,
}
),
)
[docs]
def mark_permanent_failure(self, session, archive_operation_id, exc):
"""Mark the archive operation as permanently failed."""
archive_operation = session.get(models.ArchiveOperation, archive_operation_id)
if archive_operation:
archive_operation.status = models.Status.FAILED
raw_data_package = _archive_raw_data_package(session, archive_operation)
if raw_data_package:
raw_data_package.state = models.PackageState.FAILED
archive_operation.failure_error_message = str(exc)
# Cache the latest Tier-1 breadcrumb on the row for the UI (#117).
archive_operation.error_context = self._current_error_context
logger.info(
"archive operation marked as permanently failed",
archive_operation_id=archive_operation_id,
)
_add_archive_log(
session,
archive_operation,
f"Transfer permanently failed after {archive_operation.retry_count} attempts: {str(exc)}",
)
session.commit()
redis_.publish(
"transfer:overview",
json.dumps(
{
"type": "long_term_archive_transfer_failed",
"data": archive_operation_id,
}
),
)
[docs]
def get_operation_info(self, args, kwargs):
"""Get additional context for long term archive tasks."""
if not args or len(args) == 0:
return {}
with self.session_scope() as session:
try:
archive_operation = session.get(models.ArchiveOperation, args[0])
if archive_operation:
return {
"archive_operation_id": str(archive_operation.id),
"package_id": str(archive_operation.raw_data_package_id),
"retry_count": archive_operation.retry_count,
"status": (
archive_operation.status.value
if archive_operation.status
else None
),
}
except Exception as e:
logger.error(
"error getting archive operation info",
archive_operation_id=args[0] if args else None,
error=str(e),
)
return {}
@app.task(
base=LongTermArchiveTask,
name="ccat:data_transfer:long_term_archive",
bind=True,
)
def send_data_to_long_term_archive(
self, archive_operation_id: int, session: Session = None
) -> None:
"""
Transfers a raw data package to the long term archive using dynamic queue routing.
Parameters
----------
self : celery.Task
The Celery task instance.
archive_operation_id : int
The ID of the ArchiveOperation object in the database.
Returns
-------
None
Notes
-----
- Fetches the ArchiveOperation object from the database.
- Uses dynamic queue routing based on the destination location.
- Executes the transfer command to move the data.
- Updates the ArchiveOperation status and logs in the database.
"""
logger.info(
"archive operation task started",
archive_operation_id=archive_operation_id,
)
if session is None:
with self.session_scope() as session:
return _send_archive_operation_internal(session, archive_operation_id)
else:
return _send_archive_operation_internal(session, archive_operation_id)
def _send_archive_operation_internal(
session: Session, archive_operation_id: int
) -> None:
"""Archive a raw data package on the uniform ArchiveOperation row.
Idempotent under at-least-once delivery (ADR-0004): this task may be
redelivered by the broker, re-run after a heartbeat reset, or re-scheduled by
the stuck reconciler. An already-COMPLETED operation therefore returns
immediately without re-transferring, and an existing destination copy skips
the transfer body. IN_PROGRESS is set by the base task via mark_in_progress,
NOT here (the #160 archive-local set is removed by #155); the duplicate guard
is the COMPLETED early-return plus the existing-copy no-op, never IN_PROGRESS.
"""
start_time = datetime.datetime.now()
archive_operation = _get_archive_operation(session, archive_operation_id)
# Idempotency guard: a redelivered/retried run for a finished operation must
# not move the data again.
if archive_operation.status == models.Status.COMPLETED:
logger.info(
"archive operation already completed, skipping",
archive_operation_id=archive_operation_id,
)
return
# Duplicate-killer (ADR-0004): if a physical copy already exists at the
# destination, the data is already there (e.g. a prior run completed but the
# operation row was not settled before redelivery/recovery). Skip the
# transfer body entirely and just settle the operation as COMPLETED -- running
# the body again would be a redundant re-transfer.
if _existing_destination_physical_copy(session, archive_operation):
logger.info(
"archive destination copy already present, skipping transfer body",
archive_operation_id=archive_operation_id,
)
_mark_archive_operation_successful(
session,
archive_operation,
datetime.datetime.now() - start_time,
destination_url=None,
)
return
source_url, destination_url, destination_bucket = _construct_transfer_urls(
archive_operation, session
)
_execute_transfer(
session,
source_url,
archive_operation.raw_data_package_id,
destination_url,
destination_bucket,
archive_operation,
)
end_time = datetime.datetime.now()
duration = end_time - start_time
_mark_archive_operation_successful(
session,
archive_operation,
duration,
destination_url,
)
logger.info(
"archive operation completed",
archive_operation_id=archive_operation_id,
duration_s=duration.total_seconds(),
)
redis_.publish(
"transfer:overview",
json.dumps(
{
"type": "long_term_archive_transfer_completed",
"data": archive_operation_id,
}
),
)
def _archive_raw_data_package(
session: Session, archive_operation: models.ArchiveOperation
) -> Optional[models.RawDataPackage]:
"""Resolve the RawDataPackage an ArchiveOperation archives.
ArchiveOperation carries only the FK columns (no ORM relationship), so the
package is fetched explicitly.
"""
if archive_operation.raw_data_package_id is None:
return None
return session.get(models.RawDataPackage, archive_operation.raw_data_package_id)
def _archive_destination_location(
session: Session, archive_operation: models.ArchiveOperation
) -> Optional[models.DataLocation]:
"""Resolve the destination DataLocation of an ArchiveOperation (FK-only)."""
if archive_operation.destination_location_id is None:
return None
return session.get(models.DataLocation, archive_operation.destination_location_id)
def _archive_origin_location(
session: Session, archive_operation: models.ArchiveOperation
) -> Optional[models.DataLocation]:
"""Resolve the origin DataLocation of an ArchiveOperation (FK-only)."""
if archive_operation.origin_location_id is None:
return None
return session.get(models.DataLocation, archive_operation.origin_location_id)
def _existing_destination_physical_copy(
session: Session,
archive_operation: models.ArchiveOperation,
) -> Optional[models.RawDataPackagePhysicalCopy]:
"""Return the existing copy at the operation's destination, if any.
Looks up by (raw_data_package, destination DataLocation) -- the natural key
of a physical copy. Used as the idempotency / duplicate guard (ADR-0004). The
airtight DB unique constraint on
(raw_data_package_id, destination_data_location_id), as defense-in-depth, is
deferred to #70.
"""
return (
session.query(models.RawDataPackagePhysicalCopy)
.filter(
models.RawDataPackagePhysicalCopy.raw_data_package_id
== archive_operation.raw_data_package_id,
models.RawDataPackagePhysicalCopy.data_location_id
== archive_operation.destination_location_id,
)
.first()
)
def _get_archive_operation(
session: Session, archive_operation_id: int
) -> models.ArchiveOperation:
"""Retrieve the ArchiveOperation object from the database."""
try:
archive_operation = session.get(models.ArchiveOperation, archive_operation_id)
if not archive_operation:
# Non-retryable: a missing/dangling id never reappears, so retrying
# only loops forever. A bare ValueError defaulted to retryable in
# should_retry, which is what produced the infinite archive retry.
raise OperationNotFoundError(
f"Archive operation {archive_operation_id} not found",
operation_id=archive_operation_id,
)
logger.info(
"archive operation record retrieved",
archive_operation_id=archive_operation_id,
)
return archive_operation
except Exception as e:
logger.error(
"archive operation record not found",
archive_operation_id=archive_operation_id,
error=str(e),
)
raise
def _validate_long_term_archive_path(
long_term_archive_transfer: models.LongTermArchiveTransfer,
) -> bool:
"""Validate the long term archive raw data package path."""
if (
not long_term_archive_transfer.data_archive.long_term_archive_raw_data_package_path
):
logger.error(
"LTA raw data package path not set",
lta_transfer_id=long_term_archive_transfer.id,
location_name=long_term_archive_transfer.data_archive.name,
)
return False
return True
def _construct_transfer_urls(
archive_operation: models.ArchiveOperation,
session: Session,
) -> Tuple[str, str, str]:
"""Construct the source and destination URLs using the new DataLocation system."""
# Get the source and destination locations (FK-only on ArchiveOperation).
source_location = _archive_origin_location(session, archive_operation)
destination_location = _archive_destination_location(session, archive_operation)
if not source_location or not destination_location:
raise ValueError(
"Source or destination location not set for archive operation"
)
# Get the raw data package
raw_data_package = _archive_raw_data_package(session, archive_operation)
# Construct source path based on source location type
if isinstance(source_location, models.DiskDataLocation):
source_url = safe_join(source_location.path, raw_data_package.relative_path)
elif isinstance(source_location, models.S3DataLocation):
# For source S3 locations, use the location's bucket name
source_url = (
f"s3://{source_location.bucket_name}/{raw_data_package.relative_path}"
)
else:
raise ValueError(
f"Unsupported source location type: {source_location.storage_type}"
)
# Construct destination path based on destination location type
if isinstance(destination_location, models.DiskDataLocation):
destination_url = safe_join(
destination_location.path, raw_data_package.relative_path,
)
destination_bucket = None
elif isinstance(destination_location, models.S3DataLocation):
# Use the shared function to construct the S3 key consistently
destination_url = get_s3_key_for_package(destination_location, raw_data_package)
destination_bucket = destination_location.bucket_name
else:
raise ValueError(
f"Unsupported destination location type: {destination_location.storage_type}"
)
logger.debug(
"transfer paths resolved",
source_path=source_url,
destination_path=destination_url,
source_location_type=source_location.storage_type.value,
destination_location_type=destination_location.storage_type.value,
)
return source_url, destination_url, destination_bucket
def _get_s3_metadata(
session: Session,
raw_data_package: models.RawDataPackage,
file_size: int,
) -> dict:
"""Generate comprehensive metadata for S3 upload from database models.
This function extracts metadata from the database models and constructs a dictionary
suitable for S3 object metadata. It follows a hierarchical structure for different
types of metadata (obs_, file_, archive_, etc.).
Database Relationships:
- RawDataPackage -> ObsUnit -> Source (FixedSource or other)
- RawDataPackage -> ExecutedObsUnit (direct)
- RawDataPackage -> ObsUnit -> PrimaryInstrumentModuleConfiguration -> InstrumentModule -> Instrument
- RawDataPackage -> ObsUnit -> ObservingProgram
- RawDataPackage -> ObsUnit -> SubObservingProgram
Parameters
----------
session : Session
SQLAlchemy database session
raw_data_package : models.RawDataPackage
The raw data package being uploaded
file_size : int
Size of the file in bytes
Returns
-------
dict
Dictionary of metadata key-value pairs for S3 upload
"""
# Get observation information
obs_unit = raw_data_package.obs_unit
executed_obs_unit = raw_data_package.executed_obs_unit
# Get source information
source = obs_unit.source
if isinstance(source, models.FixedSource):
ra_deg = source.ra_deg
dec_deg = source.dec_deg
target_name = source.name
else:
ra_deg = None
dec_deg = None
target_name = source.name
# Get instrument information
instrument = (
obs_unit.primary_instrument_module_configuration.instrument_module.instrument
)
# Get observing program information
program = obs_unit.observing_program
subprogram = obs_unit.sub_observing_program
# Generate a timestamp for the dataset ID if executed_obs_unit is None
timestamp = (
executed_obs_unit.start_time.strftime("%Y%m%d")
if executed_obs_unit and executed_obs_unit.start_time
else datetime.datetime.now().strftime("%Y%m%d")
)
# Log debug information
logger.debug(
"generating S3 metadata",
raw_package_id=raw_data_package.id,
obs_unit_id=obs_unit.id,
executed_obs_unit_id=executed_obs_unit.id if executed_obs_unit else None,
source_type=source.__class__.__name__,
instrument_name=instrument.name,
program_id=program.short_name,
)
# Construct metadata
metadata = {
"obs_dataset_id": f"{instrument.name}-{timestamp}-{obs_unit.id}",
"obs_telescope": instrument.telescope.name,
"obs_instrument": instrument.name,
"obs_date_obs": (
executed_obs_unit.start_time.isoformat()
if executed_obs_unit and executed_obs_unit.start_time
else None
),
# Target information
"obs_target_name": target_name,
"obs_ra_deg": str(ra_deg) if ra_deg is not None else None,
"obs_dec_deg": str(dec_deg) if dec_deg is not None else None,
# Program tracking
"obs_program_id": program.short_name,
"obs_subprogram_id": subprogram.short_name if subprogram else None,
# Database linkage
"db_obs_unit_id": str(obs_unit.id),
"db_executed_obs_unit_id": (
str(executed_obs_unit.id) if executed_obs_unit else None
),
# Data quality and status
"file_xxhash64": raw_data_package.checksum,
}
# Remove None values
metadata = {k: v for k, v in metadata.items() if v is not None}
# Log final metadata keys
logger.debug(
"s3 metadata generated",
metadata_keys=list(metadata.keys()),
metadata_count=len(metadata),
)
return metadata
def _upload_extended_metadata(
s3_client,
destination_bucket: str,
destination_url: str,
metadata: dict,
) -> bool:
"""Upload extended metadata to S3 as a separate JSON file.
Parameters
----------
s3_client : boto3.client
S3 client instance
destination_bucket : str
S3 bucket name
destination_url : str
Base destination path in S3
metadata : dict
Extended metadata to upload
Returns
-------
bool
True if upload was successful, False otherwise
"""
try:
# save to tmp file and then read and upload
with tempfile.NamedTemporaryFile(delete=False, suffix=".json") as tmp_file:
tmp_file.write(json.dumps(metadata, indent=2).encode("utf-8"))
tmp_file_path = tmp_file.name
# Read the file content for upload
with open(tmp_file_path, "rb") as f:
file_content = f.read()
# Calculate checksum from the content
metadata_checksum = base64.b64encode(
hashlib.sha256(file_content).digest()
).decode("utf-8")
# Construct metadata file path
metadata_path = f"{os.path.splitext(destination_url)[0]}_metadata.json"
# Upload metadata file with content and checksum
s3_client.put_object(
Bucket=destination_bucket,
Key=metadata_path,
Body=file_content,
ContentType="application/json",
ChecksumSHA256=metadata_checksum,
)
# Clean up temporary file
os.unlink(tmp_file_path)
logger.info(
"extended metadata upload succeeded",
metadata_path=f"s3://{destination_bucket}/{metadata_path}",
)
return True
except Exception as e:
logger.error(
"extended metadata upload failed",
error=str(e),
metadata_path=f"s3://{destination_bucket}/{metadata_path}",
)
# Clean up temporary file on error
try:
if "tmp_file_path" in locals():
os.unlink(tmp_file_path)
except Exception:
pass
return False
def _get_raw_data_package_id_from_db(session: Session, filename: str) -> int:
"""Get raw data package ID from database using filename.
Parameters
----------
session : Session
SQLAlchemy database session
filename : str
The filename to look up
Returns
-------
int
The raw data package ID
Raises
------
ValueError
If the raw data package cannot be found
"""
try:
# Get just the filename without the path
basename = os.path.basename(filename)
# Log the lookup attempt
logger.debug(
"looking up raw data package by filename", filename=basename, full_path=filename
)
# Try to find the raw data package by filename
raw_data_package = (
session.query(models.RawDataPackage)
.filter(models.RawDataPackage.name == basename)
.first()
)
if not raw_data_package:
# Try to find by relative path
raw_data_package = (
session.query(models.RawDataPackage)
.filter(models.RawDataPackage.relative_path.like(f"%{basename}"))
.first()
)
if not raw_data_package:
# Log the failed lookup
logger.error(
"raw data package not found by filename", filename=basename, full_path=filename
)
raise ValueError(f"No raw data package found for filename: {basename}")
# Log successful lookup
logger.debug(
"raw data package found", raw_package_id=raw_data_package.id, filename=basename
)
return raw_data_package.id
except Exception as e:
logger.error(
"error looking up raw data package",
error=str(e),
filename=os.path.basename(filename),
full_path=filename,
)
raise ValueError(f"Error looking up raw data package: {str(e)}")
def _execute_transfer(
session: Session,
source_url: str,
raw_data_package_id: int,
destination_url: str,
destination_bucket: str,
archive_operation: models.ArchiveOperation,
) -> None:
"""Execute the transfer between different storage types using the new DataLocation system."""
try:
start_time = time.time()
# Get the source and destination locations (FK-only on ArchiveOperation).
source_location = _archive_origin_location(session, archive_operation)
destination_location = _archive_destination_location(
session, archive_operation
)
# Determine transfer method based on location types
if isinstance(source_location, models.DiskDataLocation) and isinstance(
destination_location, models.DiskDataLocation
):
# Disk to disk transfer
_execute_disk_to_disk_transfer(source_url, destination_url)
elif isinstance(source_location, models.DiskDataLocation) and isinstance(
destination_location, models.S3DataLocation
):
logger.debug(
"disk to S3 transfer params",
source_url=source_url,
raw_package_id=raw_data_package_id,
destination_url=destination_url,
destination_bucket=destination_bucket,
)
# Disk to S3 transfer
_execute_disk_to_s3_transfer(
session,
source_url,
raw_data_package_id,
destination_url,
destination_bucket,
destination_location,
source_location.site.short_name if source_location.site else None,
source_location=source_location,
)
elif isinstance(source_location, models.S3DataLocation) and isinstance(
destination_location, models.DiskDataLocation
):
# S3 to disk transfer
_execute_s3_to_disk_transfer(source_url, destination_url)
elif isinstance(source_location, models.S3DataLocation) and isinstance(
destination_location, models.S3DataLocation
):
# S3 to S3 transfer
# _execute_s3_to_s3_transfer(source_url, destination_url, destination_bucket)
raise ValueError("S3 to S3 transfer is not supported yet")
else:
raise ValueError(
f"Unsupported transfer combination: {source_location.storage_type} to {destination_location.storage_type}"
)
# Calculate transfer metrics
end_time = time.time()
duration = end_time - start_time
# Get file size for metrics
file_size = 0
if isinstance(source_location, models.DiskDataLocation) and os.path.exists(
source_url
):
file_size = os.path.getsize(source_url)
# Send metrics to InfluxDB
if file_size > 0:
transfer_rate = (file_size / duration) / (1024 * 1024) # MB/s
metrics = HousekeepingMetrics()
try:
metrics.send_transfer_metrics(
operation="long_term_archive_transfer",
source_path=source_url,
destination_path=destination_url,
file_size=file_size,
duration=duration,
success=True,
error_message=None,
additional_fields={
"transfer_rate_mbps": transfer_rate,
},
additional_tags={
"source_location": source_location.name,
"destination_location": destination_location.name,
"transfer_id": str(archive_operation.id),
"transfer_method": f"{source_location.storage_type.value}_to_{destination_location.storage_type.value}",
},
)
except Exception as e:
logger.error(
"metrics send failed",
error=e,
transfer_id=archive_operation.id,
)
finally:
metrics.close()
except Exception as e:
logger.error(
"transfer execution failed",
error=str(e),
source_url=source_url,
destination_url=destination_url,
)
raise
def _execute_disk_to_disk_transfer(source_url: str, destination_url: str) -> None:
"""Execute disk to disk transfer using cp command."""
try:
# Create destination directory if it doesn't exist
dest_dir = os.path.dirname(destination_url)
if not os.path.exists(dest_dir):
os.makedirs(dest_dir, exist_ok=True)
# Execute copy command
cp_command = ["cp", source_url, destination_url]
logger.info("executing disk to disk transfer", command=" ".join(cp_command))
result = subprocess.run(
cp_command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
if result.returncode == 0:
logger.info(
"disk to disk transfer succeeded",
source=source_url,
destination=destination_url,
)
else:
logger.error(
"disk to disk transfer failed",
source=source_url,
destination=destination_url,
error=result.stderr,
)
except Exception as e:
logger.error(
"disk to disk transfer exception",
error=str(e),
source=source_url,
destination=destination_url,
)
raise
def _execute_disk_to_s3_transfer(
session: Session,
source_url: str,
raw_data_package_id: int,
destination_url: str,
destination_bucket: str,
destination_location: Optional[models.S3DataLocation] = None,
site_name: Optional[str] = None,
source_location: Optional[models.DiskDataLocation] = None,
) -> None:
"""Execute disk to S3 transfer.
The access method is resolved per-location (``access_method`` on the
S3DataLocation), falling back to the deployment-wide ``S3_METHOD``.
``source_location`` is threaded through so the upload functions can diagnose
a missing source against the right PhysicalCopy row (#118).
"""
method = resolve_s3_method(destination_location)
if method == "boto3":
_execute_s3_upload(
session,
source_url,
raw_data_package_id,
destination_url,
destination_bucket,
None, # No source_archive in new system
destination_location,
site_name,
source_location=source_location,
)
elif method == "coscine":
_execute_coscine_s3_upload(
session,
source_url,
raw_data_package_id,
destination_url,
destination_bucket,
destination_location,
site_name,
source_location=source_location,
)
else:
raise ValueError(f"Unsupported s3 access method: {method}")
def _execute_s3_to_disk_transfer(source_url: str, destination_url: str) -> None:
"""Execute S3 to disk transfer using aws s3 cp command."""
try:
# Create destination directory if it doesn't exist
dest_dir = os.path.dirname(destination_url)
if not os.path.exists(dest_dir):
os.makedirs(dest_dir, exist_ok=True)
# Execute S3 download command
s3_command = ["aws", "s3", "cp", source_url, destination_url]
logger.info("executing S3 to disk transfer", command=" ".join(s3_command))
result = subprocess.run(
s3_command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
if result.returncode == 0:
logger.info(
"s3 to disk transfer succeeded",
source=source_url,
destination=destination_url,
)
else:
logger.error(
"s3 to disk transfer failed",
source=source_url,
destination=destination_url,
error=result.stderr,
)
except Exception as e:
logger.error(
"s3 to disk transfer exception",
error=str(e),
source=source_url,
destination=destination_url,
)
raise
# def _execute_s3_to_s3_transfer(
# source_url: str, destination_url: str, destination_bucket: str
# ) -> None:
# """Execute S3 to S3 transfer using aws s3 cp command."""
# try:
# # Execute S3 copy command
# s3_command = ["aws", "s3", "cp", source_url, destination_url]
# logger.info("executing_s3_to_s3_transfer", command=" ".join(s3_command))
# result = subprocess.run(
# s3_command,
# stdout=subprocess.PIPE,
# stderr=subprocess.PIPE,
# text=True,
# )
# if result.returncode == 0:
# logger.info(
# "s3_to_s3_transfer_successful",
# source=source_url,
# destination=destination_url,
# )
# else:
# logger.error(
# "s3_to_s3_transfer_failed",
# source=source_url,
# destination=destination_url,
# error=result.stderr,
# )
# except Exception as e:
# logger.error(
# "s3_to_s3_transfer_exception",
# error=str(e),
# source=source_url,
# destination=destination_url,
# )
# raise
def _execute_s3_upload(
session: Session,
source_url: str,
raw_data_package_id: int,
destination_url: str,
destination_bucket: str,
source_archive: models.DataArchive,
destination_location: Optional[models.S3DataLocation] = None,
site_name: Optional[str] = None,
source_location: Optional[models.DiskDataLocation] = None,
) -> None:
"""Execute the command to upload files to S3 with MD5 verification and comprehensive metadata.
This function uploads a file to S3 with MD5 verification to ensure data integrity during transfer.
It also includes comprehensive metadata from the database to ensure proper data discovery and
provenance tracking. Additionally, it generates and uploads an extended metadata file in JSON format.
Parameters
----------
source_url : str
Path to the source file to upload
destination_url : str
Destination path in S3
destination_bucket : str
S3 bucket name
destination_location : Optional[models.S3DataLocation]
Destination S3 location for specific configuration
site_name : Optional[str]
Name of the site for credential loading
Raises
------
RuntimeError
If the S3 upload operation fails
"""
start_time = time.time()
# Guard the source read so a missing archive source surfaces as a diagnosed
# SourceMissingError rather than a bare FileNotFoundError (#118).
ensure_readable(
source_url,
side="source",
step="archive_upload_read",
host=getattr(source_location, "host", None),
diagnose=lambda: diagnose_source_missing(
session,
models.RawDataPackagePhysicalCopy,
fk_attr="raw_data_package_id",
fk_value=raw_data_package_id,
location=source_location,
)
if source_location is not None
else None,
)
# Read file data and calculate MD5
with open(source_url, "rb") as f:
file_data = f.read()
sha256_hash = hashlib.sha256(file_data).digest()
sha256_b64 = base64.b64encode(sha256_hash).decode("utf-8")
# Get file size
file_size = os.path.getsize(source_url)
# Get S3 client with location-specific configuration
s3_client = get_s3_client(destination_location, site_name)
# Query the database for package information
raw_data_package = session.query(models.RawDataPackage).get(raw_data_package_id)
if not raw_data_package:
raise ValueError(f"Raw data package {raw_data_package_id} not found")
# Get metadata from database
metadata = _get_s3_metadata(session, raw_data_package, file_size)
# Upload with MD5 verification and metadata
s3_client.put_object(
Bucket=destination_bucket,
Key=destination_url,
Body=file_data,
ChecksumSHA256=sha256_b64,
Metadata=metadata,
)
# Calculate transfer metrics
end_time = time.time()
# Generate and upload extended metadata
extended_metadata = _generate_ivoa_metadata(session, raw_data_package, file_size)
metadata_upload_success = _upload_extended_metadata(
s3_client, destination_bucket, destination_url, extended_metadata
)
if not metadata_upload_success:
logger.warning(
"extended metadata upload failed but data upload succeeded",
source=source_url,
destination=f"s3://{destination_bucket}/{destination_url}",
)
logger.info(
"s3 upload succeeded",
source=source_url,
destination=f"s3://{destination_bucket}/{destination_url}",
md5_verified=True,
metadata_keys=list(metadata.keys()),
extended_metadata_uploaded=metadata_upload_success,
)
duration = end_time - start_time
transfer_rate = (file_size / duration) / (1024 * 1024) # MB/s
transfer_metrics = {
"bytes_transferred": file_size,
"duration": duration,
"transfer_rate_mbps": transfer_rate,
}
# Send metrics to InfluxDB
metrics = HousekeepingMetrics()
try:
metrics.send_transfer_metrics(
operation="s3_upload",
source_path=source_url,
destination_path=f"s3://{destination_bucket}/{destination_url}",
file_size=transfer_metrics["bytes_transferred"],
duration=transfer_metrics["duration"],
success=True,
error_message=None,
additional_fields={
"transfer_rate_mbps": transfer_metrics["transfer_rate_mbps"],
},
additional_tags={
"source_archive": (
source_archive.short_name if source_archive else "unknown"
),
"destination_archive": "s3",
"transfer_id": str(raw_data_package_id),
"transfer_method": "s3",
},
)
except Exception as e:
logger.error("metrics send failed", error=e, transfer_id=raw_data_package_id)
finally:
metrics.close()
def _execute_coscine_s3_upload(
session: Session,
source_url: str,
raw_data_package_id: int,
destination_url: str,
destination_bucket: str,
destination_location: Optional[models.S3DataLocation] = None,
site_name: Optional[str] = None,
source_location: Optional[models.DiskDataLocation] = None,
) -> None:
"""Execute the command to upload files to S3 using Coscine with MD5 verification and comprehensive metadata.
This function uploads a file to S3 using the Coscine API with MD5 verification to ensure data integrity during transfer.
It also includes comprehensive metadata from the database to ensure proper data discovery and
provenance tracking. Additionally, it generates and uploads an extended metadata file in JSON format.
Parameters
----------
source_url : str
Path to the source file to upload
destination_url : str
Destination path in S3
destination_bucket : str
S3 bucket name
destination_location : Optional[models.S3DataLocation]
Destination S3 location for specific configuration
site_name : Optional[str]
Name of the site for credential loading
Raises
------
RuntimeError
If the S3 upload operation fails
"""
from .config.config import ccat_data_transfer_settings
start_time = time.time()
# Resolve Coscine config per-location (project/resource from the location,
# token from the env), falling back to the deployment-wide settings.
coscine_api_token, coscine_project, coscine_resource = resolve_coscine_config(
destination_location, site_name
)
# Check if COSCINE is configured
if (
not coscine_api_token
or coscine_api_token == "none"
or not coscine_project
or coscine_project == "none"
or not coscine_resource
or coscine_resource == "none"
):
raise ValueError(
"COSCINE configuration is not set. Set COSCINE_PROJECT / COSCINE_RESOURCE "
"on the data location (or in settings) and the COSCINE_API_TOKEN in the env."
)
# Guard the source read so a missing archive source surfaces as a diagnosed
# SourceMissingError rather than a bare FileNotFoundError (#118).
ensure_readable(
source_url,
side="source",
step="archive_upload_read",
host=getattr(source_location, "host", None),
diagnose=lambda: diagnose_source_missing(
session,
models.RawDataPackagePhysicalCopy,
fk_attr="raw_data_package_id",
fk_value=raw_data_package_id,
location=source_location,
)
if source_location is not None
else None,
)
# Read file data
with open(source_url, "rb") as f:
file_data = f.read()
# Note: Coscine handles integrity verification internally
# Get file size
file_size = os.path.getsize(source_url)
# Query the database for package information
raw_data_package = session.query(models.RawDataPackage).get(raw_data_package_id)
if not raw_data_package:
raise ValueError(f"Raw data package {raw_data_package_id} not found")
# Get metadata from database
metadata = _get_s3_metadata(session, raw_data_package, file_size)
# Initialize Coscine client
client = coscine.ApiClient(coscine_api_token)
project = client.project(coscine_project)
resource = project.resource(coscine_resource)
# Get metadata form and populate it
coscine_metadata = resource.metadata_form()
# Map our metadata to Coscine metadata format
# Basic metadata mapping
if "obs_target_name" in metadata:
coscine_metadata["Title"] = f"CCAT Observation: {metadata['obs_target_name']}"
else:
coscine_metadata["Title"] = (
f"CCAT Data Package: {raw_data_package.relative_path}"
)
if "obs_creator" in metadata:
coscine_metadata["Creator"] = metadata["obs_creator"]
else:
coscine_metadata["Creator"] = "CCAT Observatory"
# Add additional metadata fields only if the form supports them
if "Date" in coscine_metadata and "obs_date_obs" in metadata:
coscine_metadata["Date"] = metadata["obs_date_obs"]
if "Subject" in coscine_metadata and "obs_instrument" in metadata:
coscine_metadata["Subject"] = f"Instrument: {metadata['obs_instrument']}"
# Upload file with metadata
try:
# Create a file-like object from the file data
import io
file_obj = io.BytesIO(file_data)
# Sanitize destination path for Coscine compatibility
# Preserve readability: @NEPTUNE:Cal:BS → At_NEPTUNE-Cal-BS
destination_url = destination_url.replace("@", "At_").replace(":", "-")
# Upload the file
resource.upload(destination_url, file_obj, coscine_metadata)
# Calculate transfer metrics
end_time = time.time()
duration = end_time - start_time
transfer_rate = (file_size / duration) / (1024 * 1024) # MB/s
# Generate and upload extended metadata
extended_metadata = _generate_ivoa_metadata(
session, raw_data_package, file_size
)
# For Coscine, we'll upload the extended metadata as a separate file
metadata_upload_success = _upload_coscine_extended_metadata(
client, project, resource, destination_url, extended_metadata
)
if not metadata_upload_success:
logger.warning(
"extended metadata upload failed but data upload succeeded",
source=source_url,
destination=f"coscine://{ccat_data_transfer_settings.coscine_project}/{ccat_data_transfer_settings.coscine_resource}/{destination_url}",
)
logger.info(
"coscine S3 upload succeeded",
source=source_url,
destination=f"coscine://{ccat_data_transfer_settings.coscine_project}/{ccat_data_transfer_settings.coscine_resource}/{destination_url}",
md5_verified=True,
metadata_keys=list(metadata.keys()),
extended_metadata_uploaded=metadata_upload_success,
)
transfer_metrics = {
"bytes_transferred": file_size,
"duration": duration,
"transfer_rate_mbps": transfer_rate,
}
# Send metrics to InfluxDB
metrics = HousekeepingMetrics()
try:
metrics.send_transfer_metrics(
operation="coscine_s3_upload",
source_path=source_url,
destination_path=f"coscine://{ccat_data_transfer_settings.coscine_project}/{ccat_data_transfer_settings.coscine_resource}/{destination_url}",
file_size=transfer_metrics["bytes_transferred"],
duration=transfer_metrics["duration"],
success=True,
error_message=None,
additional_fields={
"transfer_rate_mbps": transfer_metrics["transfer_rate_mbps"],
},
additional_tags={
"source_archive": (
destination_location.name
if destination_location
else "unknown"
),
"destination_archive": "coscine",
"transfer_id": str(raw_data_package_id),
"transfer_method": "coscine",
},
)
except Exception as e:
logger.error(
"metrics send failed", error=e, transfer_id=raw_data_package_id
)
finally:
metrics.close()
except Exception as e:
logger.error(
"coscine S3 upload failed",
source=source_url,
destination=f"coscine://{ccat_data_transfer_settings.coscine_project}/{ccat_data_transfer_settings.coscine_resource}/{destination_url}",
error=str(e),
)
raise RuntimeError(f"COSCINE S3 upload failed: {str(e)}")
def _upload_coscine_extended_metadata(
client,
project,
resource,
destination_url: str,
metadata: dict,
) -> bool:
"""Upload extended metadata to Coscine as a separate JSON file.
Parameters
----------
client : coscine.ApiClient
Coscine API client instance
project : coscine.Project
Coscine project instance
resource : coscine.Resource
Coscine resource instance
destination_url : str
Base destination path
metadata : dict
Extended metadata to upload
Returns
-------
bool
True if upload was successful, False otherwise
"""
try:
# Create metadata filename
metadata_filename = f"{destination_url}_metadata.json"
# Convert metadata to JSON
metadata_json = json.dumps(metadata, indent=2)
# Create a file-like object from the JSON data
import io
metadata_file_obj = io.BytesIO(metadata_json.encode("utf-8"))
metadata_form = resource.metadata_form()
metadata_form["Title"] = "Extended Metadata"
metadata_form["Creator"] = "CCAT Observatory"
# Upload the metadata file
resource.upload(
metadata_filename,
metadata_file_obj,
metadata_form,
# {"Title": "Extended Metadata", "Creator": "CCAT Observatory"},
)
return True
except Exception as e:
logger.error(
"coscine metadata upload failed",
error=str(e),
metadata_filename=metadata_filename,
)
return False
def _mark_archive_operation_successful(
session: Session,
archive_operation: models.ArchiveOperation,
duration: datetime.timedelta,
destination_url: str,
) -> None:
"""Settle a successful archive operation and record its copy-anchored lineage.
Produces the destination archive PhysicalCopy and records the consume/produce
lineage on the ArchiveOperation (ADR-0003): it consumes the source buffer copy
and produces the destination archive copy.
"""
raw_data_package = _archive_raw_data_package(session, archive_operation)
destination_location = _archive_destination_location(session, archive_operation)
logger.info(
"archive operation marked complete",
archive_operation_id=archive_operation.id,
rel_path=raw_data_package.relative_path if raw_data_package else None,
duration_s=duration.total_seconds(),
)
# Create the physical copy record for the destination LTA location.
#
# Idempotent by construction (ADR-0004): under at-least-once delivery a prior
# run may already have created the copy, so we no-op when a copy for this
# (raw_data_package, destination DataLocation) already exists rather than
# inserting a duplicate row. This is the real duplicate-killer, and also
# guards the path where this function is reached directly (not via the
# orchestrator's pre-body check). The airtight DB unique constraint on
# (raw_data_package_id, destination_data_location_id), as defense-in-depth,
# is deferred to #70.
destination_copy = _existing_destination_physical_copy(session, archive_operation)
if destination_copy is None:
destination_copy = models.RawDataPackagePhysicalCopy(
raw_data_package=raw_data_package,
data_location=destination_location,
checksum=raw_data_package.checksum if raw_data_package else None,
verified_at=datetime.datetime.now(),
)
session.add(destination_copy)
else:
logger.info(
"physical copy already exists, skipping creation",
archive_operation_id=archive_operation.id,
raw_data_package_id=archive_operation.raw_data_package_id,
data_location_id=(
destination_location.id if destination_location else None
),
)
_record_archive_lineage(session, archive_operation, destination_copy)
archive_operation.status = models.Status.COMPLETED
if raw_data_package:
raw_data_package.state = models.PackageState.ARCHIVED
archive_operation.end_time = datetime.datetime.now()
session.commit()
redis_.publish(
"transfer:overview",
json.dumps(
{
"type": "long_term_archive_transfer_completed",
"data": archive_operation.id,
}
),
)
_add_archive_log(
session,
archive_operation,
f"Transfer successful to long term archive in {duration}",
)
def _record_archive_lineage(
session: Session,
archive_operation: models.ArchiveOperation,
destination_copy: models.RawDataPackagePhysicalCopy,
) -> None:
"""Record the operation's consume/produce PhysicalCopy lineage (ADR-0003).
The archive operation consumes the source buffer copy (the package copy at
the origin location) and produces the destination archive copy. Membership is
set-idempotent so a redelivered run does not duplicate lineage edges.
"""
if destination_copy not in archive_operation.produced_copies:
archive_operation.produced_copies.append(destination_copy)
source_copy = (
session.query(models.RawDataPackagePhysicalCopy)
.filter(
models.RawDataPackagePhysicalCopy.raw_data_package_id
== archive_operation.raw_data_package_id,
models.RawDataPackagePhysicalCopy.data_location_id
== archive_operation.origin_location_id,
)
.first()
)
if source_copy is not None and source_copy not in archive_operation.consumed_copies:
archive_operation.consumed_copies.append(source_copy)
def _add_archive_log(
session: Session,
archive_operation: models.ArchiveOperation,
log_message: str,
) -> None:
"""Emit a structured log line for the archive operation.
The legacy ``LongTermArchiveTransferLog`` rows are tied to the legacy transfer
FK; the durable failure trail now lives in ``OperationFailureEvent`` (#117),
so the operation lifecycle logs through the structured logger rather than the
legacy log table.
"""
logger.info(
"archive operation log",
archive_operation_id=archive_operation.id,
log=log_message,
)
[docs]
def transfer_raw_data_packages_to_long_term_archive(
verbose: bool = False, site_name: Optional[str] = None
) -> None:
"""
Schedule long term archive transfer tasks for pending raw data packages using the new DataLocation system.
Args:
verbose (bool): If True, sets logging to DEBUG level. Defaults to False.
site_name (Optional[str]): If provided, only schedules transfers for the specified site.
Raises:
SQLAlchemyError: If there's an issue with database operations.
"""
db = DatabaseConnection()
session, _ = db.get_connection()
# Liveness probe for the stuck-operation backstop (#161); reuses the module
# Redis client that _mark_archive_operation_successful publishes through.
task_state_manager = TaskStateManager(redis_)
# Get all sites with LTA locations
lta_sites = _get_sites_with_lta_locations(session, site_name)
logger.debug(
"LTA sites found",
sites=[site.short_name for site in lta_sites],
count=len(lta_sites),
)
for site in lta_sites:
# Get all LTA locations for this site
lta_locations = _get_lta_locations_for_site(session, site)
for lta_location in lta_locations:
# Find pending existing archive operations for this LTA location
pending_existing_operations = (
_get_pending_existing_operations_to_lta_location(session, lta_location)
)
if len(pending_existing_operations) > 0:
for operation in pending_existing_operations:
_schedule_archive_task(session, operation)
# Backstop recovery for genuinely dead operations (#161). The primary
# dead-worker path is task_monitor heartbeat recovery; this only
# surfaces operations with no live heartbeat AND a start_time past the
# reset floor (default = broker visibility_timeout), so it never races
# the broker's redelivery window nor false-resets a healthy
# long-running archive (#151).
stuck_operations = _get_stuck_archive_operations(
session, lta_location, task_state_manager
)
if len(stuck_operations) > 0:
logger.info(
"recovering stuck archive operations",
site=site.short_name,
location=lta_location.name,
stuck_count=len(stuck_operations),
)
for operation in stuck_operations:
previous_status = operation.status.value
logger.warning(
"resetting stuck archive operation (no live heartbeat past floor)",
lta_transfer_id=operation.id,
location=lta_location.name,
previous_status=previous_status,
retry_count=operation.retry_count,
)
operation.status = models.Status.PENDING
operation.retry_count += 1
_add_archive_log(
session,
operation,
f"Transfer stuck in {previous_status}, resetting for retry",
)
session.commit()
_schedule_archive_task(session, operation)
# Find new archive operations that need to be created for this location
pending_new_packages = _get_pending_new_transfers_to_lta_location(
session, lta_location
)
if len(pending_new_packages) > 0:
logger.info(
"new archive operations pending",
site=site.short_name,
location=lta_location.name,
package_count=len(pending_new_packages),
)
for package in pending_new_packages:
archive_operation = _create_archive_operation(
session, package, lta_location
)
_schedule_archive_task(session, archive_operation)
session.commit()
redis_.publish(
"transfer:overview",
json.dumps(
{
"type": "long_term_archive_transfer_scheduled",
}
),
)
def _get_pending_existing_operations_to_lta_location(
session: Session, lta_location: models.DataLocation
) -> list:
return (
session.query(models.ArchiveOperation)
.filter(
models.ArchiveOperation.destination_location_id == lta_location.id,
models.ArchiveOperation.status == models.Status.PENDING,
)
.all()
)
def _get_stuck_archive_operations(
session: Session,
lta_location: models.DataLocation,
task_state_manager: TaskStateManager,
stuck_reset_floor_seconds: Optional[int] = None,
) -> list:
"""
Retrieve archive operations that are genuinely stuck with no live worker (#161).
This is the BACKSTOP, not the primary dead-worker path. ``task_monitor``
heartbeat recovery (task_monitor_service.py) is the primary mechanism: it
resets a registered-but-stale task as soon as its heartbeat lapses. This
reconciler only catches operations that path can miss -- tasks that were
never registered in Redis, or where the DB and Redis have diverged.
An operation is reset SCHEDULED/IN_PROGRESS -> PENDING only when BOTH:
(a) no sibling task is alive for the operation -- ``is_operation_alive``
is False (a live heartbeat means a worker is still on it; resetting it
would be the #151 false-positive that kills a healthy long-running
archive on a blind DB clock); AND
(b) ``start_time`` predates the reset floor. The floor defaults to the
broker ``visibility_timeout`` (3600s, see setup_celery_app.py) so this
path can never race the broker's own redelivery window: inside the
floor the broker may still redeliver the original task.
The ``retry_count < 3`` guard caps retries.
Args:
session: The database session.
lta_location: The LTA location to check.
task_state_manager: Provides the Redis-backed liveness probe
(``is_operation_alive``).
stuck_reset_floor_seconds: Lower time bound before an operation may be
reset. Defaults to ``TASK_RECOVERY.STUCK_RESET_FLOOR_SECONDS``.
Returns:
list: ArchiveOperation objects that appear stuck and dead.
"""
from .config.config import ccat_data_transfer_settings
if stuck_reset_floor_seconds is None:
stuck_reset_floor_seconds = (
ccat_data_transfer_settings.TASK_RECOVERY.STUCK_RESET_FLOOR_SECONDS
)
cutoff = datetime.datetime.now() - datetime.timedelta(
seconds=stuck_reset_floor_seconds
)
candidates = (
session.query(models.ArchiveOperation)
.filter(
models.ArchiveOperation.destination_location_id == lta_location.id,
models.ArchiveOperation.status.in_(
[models.Status.SCHEDULED, models.Status.IN_PROGRESS]
),
models.ArchiveOperation.start_time < cutoff,
models.ArchiveOperation.retry_count < 3,
)
.all()
)
# Heartbeat gate: drop any candidate whose worker is still beating, so the
# backstop never overrides a live task that task_monitor would not touch.
# Keyed by OperationKind.ARCHIVE (identity), whose value is the unchanged
# Redis key fragment — no literal op-kind string here (ADR-0003).
return [
operation
for operation in candidates
if not task_state_manager.is_operation_alive(
models.OperationKind.ARCHIVE, operation.id
)
]
def _get_sites_with_lta_locations(
session: Session, site_name: Optional[str] = None
) -> list:
"""Get all sites that have LTA locations."""
query = (
session.query(models.Site)
.join(models.DataLocation)
.filter(
models.DataLocation.location_type == models.LocationType.LONG_TERM_ARCHIVE,
models.DataLocation.active == True, # noqa: E712
)
.distinct()
)
if site_name:
query = query.filter(models.Site.short_name == site_name)
return query.all()
def _get_lta_locations_for_site(session: Session, site: models.Site) -> list:
"""Get all LTA locations for a specific site."""
return (
session.query(models.DataLocation)
.filter(
models.DataLocation.site_id == site.id,
models.DataLocation.location_type == models.LocationType.LONG_TERM_ARCHIVE,
models.DataLocation.active == True, # noqa: E712
)
.all()
)
def _get_pending_new_transfers_to_lta_location(
session: Session, lta_location: models.DataLocation
) -> list:
"""Get raw data packages that need to be transferred to a specific LTA location."""
# Find packages that are in buffer locations and haven't been transferred to this LTA location yet
# Debug: Log the LTA location details
logger.debug(
"checking LTA location",
location_id=lta_location.id,
location=lta_location.name,
site_id=lta_location.site_id,
site=lta_location.site.short_name,
)
# Debug: Check what buffer locations exist for this site
buffer_locations = (
session.query(models.DataLocation.id, models.DataLocation.name)
.filter(
models.DataLocation.location_type == models.LocationType.BUFFER,
models.DataLocation.active == True, # noqa: E712
models.DataLocation.site_id == lta_location.site_id,
)
.all()
)
logger.debug(
"buffer locations for site",
site_id=lta_location.site_id,
site=lta_location.site.short_name,
buffer_locations=[{"id": loc.id, "name": loc.name} for loc in buffer_locations],
)
# Check existing archive operations for this destination so a package already
# being archived here is not re-created.
existing_operations = (
session.query(models.ArchiveOperation.raw_data_package_id)
.filter(
models.ArchiveOperation.destination_location_id == lta_location.id,
)
.all()
)
existing_package_ids = [o.raw_data_package_id for o in existing_operations]
logger.debug(
"existing archive operations for destination",
destination_location_id=lta_location.id,
location=lta_location.name,
existing_package_count=len(existing_package_ids),
existing_package_ids=existing_package_ids,
)
# A package present in several buffer locations is returned once per matching
# physical copy, so dedupe by id in Python rather than SELECT DISTINCT (Postgres
# cannot DISTINCT the json error_context column on this model — see
# deduplicate_by_id).
pending_packages = deduplicate_by_id(
session.query(models.RawDataPackage)
.join(models.RawDataPackagePhysicalCopy)
.filter(
models.RawDataPackagePhysicalCopy.data_location_id.in_(
[loc.id for loc in buffer_locations]
),
models.RawDataPackagePhysicalCopy.status
== models.PhysicalCopyStatus.PRESENT,
~models.RawDataPackage.id.in_(existing_package_ids),
)
.all()
)
logger.info(
"pending packages for LTA location",
location=lta_location.name,
site=lta_location.site.short_name,
package_count=len(pending_packages),
)
# Debug: Log details of found packages
if pending_packages:
package_details = []
for package in pending_packages:
physical_copies = [
{
"location_id": pc.data_location_id,
"location_name": pc.data_location.name,
"status": pc.status.value,
}
for pc in package.physical_copies
]
package_details.append(
{
"id": package.id,
"name": package.name,
"physical_copies": physical_copies,
}
)
logger.debug(
"pending package details",
packages=package_details,
)
return pending_packages
def _create_archive_operation(
session: Session, package: models.RawDataPackage, lta_location: models.DataLocation
) -> models.ArchiveOperation:
"""Create a new archive operation for a package to a specific LTA location."""
# Find the source buffer location for this package
source_physical_copy = (
session.query(models.RawDataPackagePhysicalCopy)
.join(models.DataLocation)
.filter(
models.RawDataPackagePhysicalCopy.raw_data_package_id == package.id,
models.DataLocation.location_type == models.LocationType.BUFFER,
models.DataLocation.site_id == lta_location.site_id,
models.DataLocation.active == True, # noqa: E712
models.RawDataPackagePhysicalCopy.status
== models.PhysicalCopyStatus.PRESENT,
)
.first()
)
if not source_physical_copy:
raise ValueError(f"No source buffer found for package {package.id}")
archive_operation = models.ArchiveOperation(
raw_data_package_id=package.id,
origin_location_id=source_physical_copy.data_location_id,
destination_location_id=lta_location.id,
status=models.Status.PENDING,
)
session.add(archive_operation)
session.flush()
session.commit()
redis_.publish(
"transfer:overview",
json.dumps(
{
"type": "long_term_archive_transfer_created",
"data": archive_operation.id,
}
),
)
return archive_operation
def _max_retries_reached(
archive_operation: models.ArchiveOperation,
) -> bool:
return archive_operation.retry_count >= 3
def _schedule_archive_task(
session: Session,
archive_operation: models.ArchiveOperation,
) -> None:
"""Schedule an archive operation task using dynamic queue routing."""
destination_location = _archive_destination_location(session, archive_operation)
try:
# Use dynamic queue routing based on the destination location
queue_name = route_task_by_location(
OperationType.LONG_TERM_ARCHIVE_TRANSFER,
destination_location,
)
logger.info(
"scheduling archive task",
archive_operation_id=archive_operation.id,
destination_location=(
destination_location.name if destination_location else None
),
queue=queue_name,
)
# Schedule the task with the appropriate queue
send_data_to_long_term_archive.apply_async(
args=[archive_operation.id],
queue=queue_name,
)
archive_operation.status = models.Status.SCHEDULED
archive_operation.start_time = datetime.datetime.now()
except Exception as e:
logger.error(
"archive task scheduling failed",
archive_operation_id=archive_operation.id,
error=str(e),
)
raise ScheduleError(
f"Error scheduling archive operation task: {str(e)}"
)
finally:
session.commit()
logger.info(
"archive task scheduled",
archive_operation_id=archive_operation.id,
queue=queue_name,
)
redis_.publish(
"transfer:overview",
json.dumps(
{
"type": "long_term_archive_transfer_scheduled",
"data": archive_operation.id,
"queue": queue_name,
}
),
)
def _generate_ivoa_metadata(
session: Session,
raw_data_package: models.RawDataPackage,
file_size: int,
) -> dict:
"""Generate IVOA-compatible extended metadata for a raw data package.
This function creates a comprehensive metadata document that follows IVOA standards
while maintaining flexibility for different instruments and observatories. It combines
data from the core database models with additional metadata stored in the
RawDataPackageMetadata table.
Parameters
----------
session : Session
SQLAlchemy database session
raw_data_package : models.RawDataPackage
The raw data package being uploaded
file_size : int
Size of the file in bytes
Returns
-------
dict
IVOA-compatible metadata document
Notes
-----
The metadata structure follows IVOA standards while maintaining flexibility:
1. Core identifiers follow IVOA URI conventions
2. Facility information is standardized but extensible
3. Instrument-specific metadata is stored in RawDataPackageMetadata
4. All numeric values are stored with units
5. Timestamps are in ISO 8601 format with timezone
"""
# Get observation information
obs_unit = raw_data_package.obs_unit
executed_obs_unit = raw_data_package.executed_obs_unit
# Get source information
source = obs_unit.source
if isinstance(source, models.FixedSource):
ra_deg = source.ra_deg
dec_deg = source.dec_deg
target_name = source.name
else:
ra_deg = None
dec_deg = None
target_name = source.name
# Get instrument information
instrument = (
obs_unit.primary_instrument_module_configuration.instrument_module.instrument
)
# Get observing program information
program = obs_unit.observing_program
subprogram = obs_unit.sub_observing_program
# Get additional metadata
additional_metadata = raw_data_package.package_metadata
# Construct base metadata
metadata = {
# Core identifiers (IVOA compatible)
"id": {
"dataset_id": f"{instrument.name}-{executed_obs_unit.start_time.strftime('%Y%m%d')}-{obs_unit.id}",
"obs_unit_id": str(obs_unit.id),
"executed_obs_unit_id": (
str(executed_obs_unit.id) if executed_obs_unit else None
),
"publisher_did": f"ivo://org.ccat-p/raw/{instrument.name}-{executed_obs_unit.start_time.strftime('%Y%m%d')}-{obs_unit.id}",
"ivoa_collection": "ivo://org.ccat-p/collection/raw-observations",
},
# Facility information (standardized)
"facility": {
"observatory": {
"name": instrument.telescope.observatory.name,
"altitude_m": instrument.telescope.alt_m,
"longitude_deg": instrument.telescope.lon_deg,
"latitude_deg": instrument.telescope.lat_deg,
},
"telescope": {
"name": instrument.telescope.name,
"longitude_deg": instrument.telescope.lon_deg,
"latitude_deg": instrument.telescope.lat_deg,
"altitude_m": instrument.telescope.alt_m,
},
"instrument": {
"name": instrument.name,
"type": instrument.instrument_type,
"description": instrument.description,
"modules": [
{
"name": module.instrument_module.name,
}
for module in obs_unit.instrument_module_configurations
],
},
},
# Target information
"target": {
"name": target_name,
"type": source.__class__.__name__.lower(),
"coordinates": {
"ra": {
"degrees": ra_deg,
"sexagesimal": (
_degrees_to_sexagesimal(ra_deg) if ra_deg is not None else None
),
},
"dec": {
"degrees": dec_deg,
"sexagesimal": (
_degrees_to_sexagesimal(dec_deg)
if dec_deg is not None
else None
),
},
"frame": "ICRS",
"epoch": "J2000",
},
},
# Observation configuration (IVOA compatible)
"observation": {
"program": {
"id": program.short_name,
"name": program.name,
# "pi": program.pi_name,
# "pi_affiliation": program.pi_affiliation,
# "proposal_id": program.proposal_id,
},
"subprogram": {
"id": subprogram.short_name if subprogram else None,
"name": subprogram.name if subprogram else None,
},
"timing": {
"start_time": (
executed_obs_unit.start_time.isoformat()
if executed_obs_unit
else None
),
"end_time": (
executed_obs_unit.end_time.isoformat()
if executed_obs_unit
else None
),
"duration_s": (
(
executed_obs_unit.end_time - executed_obs_unit.start_time
).total_seconds()
if executed_obs_unit
else None
),
},
},
# Contents information
"contents": {
"file_count": len(raw_data_package.raw_data_files),
"size_bytes": file_size,
"files": [
{
"filename": file.name,
"type": file.file_type,
"size_bytes": file.size,
"checksum": {"algorithm": "xxh64", "value": file.checksum},
}
for file in raw_data_package.raw_data_files
],
},
# Data quality assessment
"quality": {
"status": executed_obs_unit.status if executed_obs_unit else None,
},
# Processing information
"processing": {
"level": "raw",
"pipeline": {
"name": instrument.name,
"version": "1.0", # Default version since it's not in the model
},
},
# Data rights and access information
# "access": {
# "policy": program.data_policy,
# },
# Full provenance chain
# "provenance": {
# "creator": instrument.name, # Use instrument name since data_acquisition_system doesn't exist
# "created": raw_data_package.created_at.isoformat(),
# "contributors": (
# [
# {
# "name": executed_obs_unit.observer_name,
# "role": "Observer",
# "affiliation": executed_obs_unit.observer_affiliation,
# }
# ]
# if executed_obs_unit
# else []
# ),
# },
}
# Add additional metadata if available
if additional_metadata:
# Add instrument-specific metadata
if additional_metadata.instrument_specific:
metadata["instrument_specific"] = additional_metadata.instrument_specific
# Add quality metrics
if additional_metadata.quality_metrics:
metadata["quality"].update(additional_metadata.quality_metrics)
# Add extended provenance
if additional_metadata.provenance:
metadata["provenance"].update(additional_metadata.provenance)
# Add custom metadata
if additional_metadata.custom_metadata:
metadata["custom"] = additional_metadata.custom_metadata
# Remove None values
metadata = _remove_none_values(metadata)
return metadata
def _degrees_to_sexagesimal(degrees: float) -> str:
"""Convert decimal degrees to sexagesimal format."""
if degrees is None:
return None
# Handle negative degrees for declination
sign = "-" if degrees < 0 else "+"
degrees = abs(degrees)
# Convert to hours/degrees, minutes, seconds
hours = int(degrees)
minutes = int((degrees - hours) * 60)
seconds = (degrees - hours - minutes / 60) * 3600
return f"{sign}{hours:02d}h{minutes:02d}m{seconds:06.3f}s"
def _remove_none_values(d: dict) -> dict:
"""Recursively remove None values from a dictionary."""
if not isinstance(d, dict):
return d
return {
k: _remove_none_values(v)
for k, v in d.items()
if v is not None and (not isinstance(v, dict) or _remove_none_values(v))
}
