diff --git a/docs/dsh/dsh.md b/docs/dsh/dsh.md new file mode 100644 index 0000000..e394881 --- /dev/null +++ b/docs/dsh/dsh.md @@ -0,0 +1,41 @@ +# DSH + +Only files from the `pseudonymised` directory are uploaded. +There are safeguards to avoid accidentally uploading from any other directory. + +## DSH FTPS + +### Write-only uploader accounts + +See slab article on +[how to configure the uploader accounts](https://uclh.slab.com/posts/ftps-dsh-uploads-9otokl8x). + +### Notifications + +One email notification is sent from the DSH per uploaded file. +We need to upload ~hundreds every day, therefore we use a temporary TAR file so that +all our parquets are uploaded in one go. + +The TAR file is named according to the *time of upload*, but the file structure within it +is done according to the event times of the data. + +Example output of `tar tvf`: +``` +-rw-r--r-- 0 root root 10622 24 Aug 17:17 2024-09-12/2024-09-12.4e121edfa3d75b935975bdf2db2c32e229ab3c2764873b506b3fec192bd0b8ec.1570.noCh.cmH2O.parquet +-rw-r--r-- 0 root root 10982 24 Aug 17:17 2024-09-12/2024-09-12.6aae1d263b6029b2344750c9e56a2ccadbd2bf6b08bd0fb6469f4274d96fbb3d.1408.noCh.s.parquet +... +``` + +Naming by upload time means that subsequently uploaded TAR files +will never overwrite previous ones. +This allows for incremental uploads; that is, the addition of extra data +(eg. new variables, new patients) +for dates that have already had an upload in the past. +*However*, the extracted files will clash in name, as the names +of the parquets within are anchored to the original event date. +It would be the job of a future DSH extractor script to do the right thing here. +Eg. to have a rule that extracts from later uploads always take precedence. +See [issue #84](https://github.com/SAFEHR-data/waveform-controller/issues/84) . + +The uploaded file name is stored in JSON on the GAE in the daily uploaded sentinel file: +eg. `waveform-export/ftps-logs/2024-09-12/2024-09-12.uploaded.json` diff --git a/exporter-scripts/scheduled-script.sh b/exporter-scripts/scheduled-script.sh index 7692e5d..5b25d99 100755 --- a/exporter-scripts/scheduled-script.sh +++ b/exporter-scripts/scheduled-script.sh @@ -29,6 +29,7 @@ ONLY_USE_CSV_FROM_YESTERDAY="${ONLY_USE_CSV_FROM_YESTERDAY:-True}" PROCESS_CSV_FROM_DATE="${PROCESS_CSV_FROM_DATE:-'[0-9]'}" set +e snakemake --snakefile /app/src/pipeline/Snakefile \ + --resources ftps_server=1 \ --cores "$SNAKEMAKE_CORES" \ --until "$SNAKEMAKE_RULE_UNTIL" \ --config CSV_AGE_THRESHOLD_MINUTES="${CSV_AGE_THRESHOLD_MINUTES}" ONLY_USE_CSV_FROM_YESTERDAY="${ONLY_USE_CSV_FROM_YESTERDAY}" PROCESS_CSV_FROM_DATE="${PROCESS_CSV_FROM_DATE}"\ diff --git a/src/exporter/ftps.py b/src/exporter/ftps.py index 3f6a040..ca0e088 100644 --- a/src/exporter/ftps.py +++ b/src/exporter/ftps.py @@ -1,13 +1,20 @@ import argparse +import json import logging -import os +import tarfile from pathlib import Path +from tempfile import NamedTemporaryFile +from time import perf_counter +from typing import Any -import settings -from core.uploader._ftps import _connect_to_ftp, _create_and_set_as_cwd +from core.uploader._ftps import _connect_to_ftp, _create_and_set_as_cwd_multi_path +import settings +import telemetry from locations import WAVEFORM_PSEUDONYMISED_PARQUET +logger = logging.getLogger(__name__) + def do_upload_cli(): parser = argparse.ArgumentParser() @@ -17,46 +24,128 @@ def do_upload_cli(): help="file to upload relative to pseudonymised folder", ) args = parser.parse_args() - do_upload(args.file_to_upload) + do_upload_multiple(args.file_to_upload) -def do_upload(abs_file_to_upload: Path): +def do_upload_multiple_with_telemetry( + file_list: list[Path], remote_tar_filename: str, wc_date: str +): + start_perf = perf_counter() + logger.info( + "Calling do_upload_multiple to create temp tar file %s", remote_tar_filename + ) + attrs = { + "obs_date": wc_date, + } + try: + do_upload_multiple(file_list, remote_tar_filename) + except Exception as e: + attrs["error.type"] = str(type(e)) + logger.exception( + "FTPS upload failed for remote filename %s", remote_tar_filename, exc_info=e + ) + raise + finally: + perf_time = perf_counter() - start_perf + telemetry.ftps_uploaded.add(1, attributes=attrs) + telemetry.ftps_time_taken.record(perf_time) + return perf_time + + +def do_upload_multiple( + abs_files_to_upload: list[Path], remote_tar_filename: str +) -> None: """We need to ensure that a user cannot accidentally ask for a file to be uploaded unless it's under the correct directory that we know contains pseudonymised data.""" - logger = logging.getLogger(__name__) # Keep things simple, paths must be absolute - if not abs_file_to_upload.is_absolute(): - raise ValueError("File must be relative to pseudonymised folder") - # Even an absolute path may contain a ".." or a symlink. Fully resolve so we - # know what we are dealing with. - file_to_upload = abs_file_to_upload.resolve() - # Check the file is still under the "safe" directory for upload. - if not file_to_upload.is_relative_to(WAVEFORM_PSEUDONYMISED_PARQUET): - raise ValueError( - f"File {file_to_upload} must be under {WAVEFORM_PSEUDONYMISED_PARQUET}. " - f"If this is unexpected, maybe you are using symlinks or '..' in the path?" - ) - if not file_to_upload.exists(): - raise ValueError(f"File {file_to_upload} does not exist") - logger.info( - "Connecting to FTPS server %s:%s, with username %s", - settings.FTPS_HOST, - settings.FTPS_PORT, - settings.FTPS_USERNAME, + rel_norm_files_to_upload = [] + for abs_file in abs_files_to_upload: + if not abs_file.is_absolute(): + raise ValueError("File must be relative to pseudonymised folder") + # Even an absolute path may contain a ".." or a symlink. Fully resolve so we + # know what we are dealing with. + norm_file = abs_file.resolve() + # Check the file is still under the "safe" directory for upload. + try: + rel_norm_file = norm_file.relative_to(WAVEFORM_PSEUDONYMISED_PARQUET) + except ValueError as e: + raise ValueError( + f"File {norm_file} must be under {WAVEFORM_PSEUDONYMISED_PARQUET}. " + f"If this is unexpected, maybe you are using symlinks or '..' in the path?" + ) from e + if not norm_file.exists(): + raise ValueError(f"File {norm_file} does not exist") + rel_norm_files_to_upload.append(rel_norm_file) + # We get one notification email per file uploaded, so tar it up to reduce this. + # Use a directory under WAVEFORM_PSEUDONYMISED_PARQUET so we keep all the pseudon data + # in one place. + tmp_tar_dir = WAVEFORM_PSEUDONYMISED_PARQUET / "tmp_tar" + tmp_tar_dir.mkdir(exist_ok=True) + remote_project_dir = ( + Path("waveform-export") / settings.INSTANCE_NAME / "pseudonymised" ) - ftp = _connect_to_ftp( - settings.FTPS_HOST, - settings.FTPS_PORT, - settings.FTPS_USERNAME, - settings.FTPS_PASSWORD, + logger.info( + "tmp_tar_dir: %s,\nremote_project_dir = %s", tmp_tar_dir, remote_project_dir ) - remote_project_dir = str(Path("waveform-export") / settings.INSTANCE_NAME) - _create_and_set_as_cwd(ftp, remote_project_dir) - remote_filename = os.path.basename(file_to_upload) - command = f"STOR {remote_filename}" - logger.info("Uploading file %s", file_to_upload) - with open(file_to_upload, "rb") as file_to_upload_fh: - ftp.storbinary(command, file_to_upload_fh) - print("Directory listing: ") - ftp.dir() - ftp.quit() + with NamedTemporaryFile( + dir=tmp_tar_dir, delete_on_close=False, delete=False + ) as temp_tar_file_path: + logger.info("Making temp tarfile: %s", temp_tar_file_path.name) + with tarfile.TarFile(fileobj=temp_tar_file_path, mode="w") as tar_file: + for file_to_upload in rel_norm_files_to_upload: + tar_file.add( + WAVEFORM_PSEUDONYMISED_PARQUET / file_to_upload, + arcname=file_to_upload, + ) + # tar writer has finished writing, but flush to disk and seek to beginning of file + temp_tar_file_path.flush() + temp_tar_file_path.seek(0) + logger.info( + "Connecting to FTPS server %s:%s, with username %s", + settings.FTPS_HOST, + settings.FTPS_PORT, + settings.FTPS_USERNAME, + ) + ftp = _connect_to_ftp( + settings.FTPS_HOST, + settings.FTPS_PORT, + settings.FTPS_USERNAME, + settings.FTPS_PASSWORD, + ) + _create_and_set_as_cwd_multi_path(ftp, remote_project_dir) + command = f"STOR {remote_tar_filename}" + tar_file_size = Path(temp_tar_file_path.name).stat().st_size + logger.info( + "Uploading temp tarfile as %s in remote dir %s (%s bytes)", + remote_tar_filename, + remote_project_dir, + tar_file_size, + ) + resp_code = ftp.storbinary(command, temp_tar_file_path) + # Log but don't check the response code; rely on raising one + # of the ftplib exceptions to detect errors + logger.info("FTP response code: %s", resp_code) + # I wanted to upload with a ".part" suffix, then rename to remove the + # suffix, to make it very clear to the DSH end that the file transfer completed. + # However, renaming results in error_perm (550 Permission denied), presumably because + # of the write-only policy. + print("Directory listing: ") + ftp.dir() + ftp.quit() + + +def write_ftps_sentinel( + overall_stats_dict: dict[str, Any], + sentinel_file: Path, + uploaded_files: list[Path], +): + sentinel_data = { + "overall": overall_stats_dict, + "uploaded_files": uploaded_files, + } + with open(sentinel_file, "w") as fh: + json.dump( + sentinel_data, + fh, + indent=0, + ) diff --git a/src/locations.py b/src/locations.py index bb15847..a3c867f 100644 --- a/src/locations.py +++ b/src/locations.py @@ -21,6 +21,10 @@ ) HASH_LOOKUP_JSON_REL = Path("{date}/{date}.hashes.json") HASH_LOOKUP_JSON = WAVEFORM_HASH_LOOKUPS / HASH_LOOKUP_JSON_REL +ALL_UPLOADED_JSON_REL = Path("{date}/{date}.uploaded.json") +ALL_UPLOADED_JSON = WAVEFORM_FTPS_LOGS / ALL_UPLOADED_JSON_REL +ALL_FTPS_LOG_REL = Path("{date}/{date}.ftps.log") +ALL_FTPS_LOG = WAVEFORM_FTPS_LOGS / ALL_FTPS_LOG_REL def make_file_name(template: str, subs: dict[str, str]): diff --git a/src/pipeline/Snakefile b/src/pipeline/Snakefile index 97965a4..3876b86 100644 --- a/src/pipeline/Snakefile +++ b/src/pipeline/Snakefile @@ -1,25 +1,24 @@ -import json -import time +import shutil from datetime import datetime, timedelta, timezone from pathlib import Path from exporter.daily_summary import make_daily_hash_summary -from exporter.ftps import do_upload +from exporter.ftps import write_ftps_sentinel, do_upload_multiple_with_telemetry from locations import ( WAVEFORM_ORIGINAL_CSV, WAVEFORM_SNAKEMAKE_LOGS, WAVEFORM_PSEUDONYMISED_PARQUET, - WAVEFORM_FTPS_LOGS, HASH_LOOKUP_JSON, HASH_LOOKUP_JSON_REL, FILE_STEM_PATTERN, FILE_STEM_PATTERN_HASHED, make_file_name, + ALL_UPLOADED_JSON, + ALL_FTPS_LOG, ) +from pipeline.utils import config_bool, determine_eventual_outputs, timestamp_for_paths from pseudon.pseudon import csv_to_parquets -from utils import config_bool, determine_eventual_outputs, report_ftp_upload - # How long before we assume that no more data will be written to the file, and # that we can process it. @@ -45,7 +44,7 @@ PROCESS_CSV_FROM_DATE = str(config['PROCESS_CSV_FROM_DATE']) # be fed into snakemake. all_outputs, hash_to_csn = determine_eventual_outputs(CSV_AGE_THRESHOLD_MINUTES, ONLY_USE_CSV_FROM_YESTERDAY, PROCESS_CSV_FROM_DATE) -ALL_FTPS_UPLOADED = [ao.get_ftps_uploaded_file() for ao in all_outputs] +ALL_FTPS_UPLOADED = sorted({ao.get_ftps_uploaded_all_file() for ao in all_outputs}) ALL_DAILY_HASH_LOOKUPS = sorted({ao.get_daily_hash_lookup() for ao in all_outputs}) def configure_file_logging(log_file): @@ -137,32 +136,46 @@ rule daily_hash_lookup: make_daily_hash_summary(daily_files, output.hash_lookup_json) -rule send_ftps: +rule send_all_ftps: + resources: + # We want to be connected to the FTPS server only once at any one time, + # so consume our one and only instance of this resource + ftps_server=1 input: - WAVEFORM_PSEUDONYMISED_PARQUET / (FILE_STEM_PATTERN_HASHED + ".parquet") + pseudonymised_parquets = pseudonymised_parquet_files_for_date output: - # sentinel file - WAVEFORM_FTPS_LOGS / (FILE_STEM_PATTERN_HASHED + ".ftps.uploaded.json") + # Files are uploaded as a temporary tar file, so there is only one sentinel, + # which contains some data about the files uploaded + overall_uploaded_sentinel = ALL_UPLOADED_JSON log: - WAVEFORM_FTPS_LOGS / (FILE_STEM_PATTERN_HASHED + ".ftps.log") + ftps_log_file = ALL_FTPS_LOG run: - start_perf = time.perf_counter() + logger = configure_file_logging(log.ftps_log_file) start_timestamp = datetime.now(timezone.utc).isoformat() - logger = configure_file_logging(log[0]) - logger.info("Calling do_upload to upload file %s", input[0]) - do_upload(Path(input[0])) - report_ftp_upload() - end_perf = time.perf_counter() + file_safe_timestamp = timestamp_for_paths() + remote_tar_filename = f"upload.{file_safe_timestamp}.tar" + + perf_time = do_upload_multiple_with_telemetry([Path(i) for i in input], remote_tar_filename, wildcards.date) + end_timestamp = datetime.now(timezone.utc).isoformat() - # now that we have success, create the sentinel file - with open(output[0], "w") as fh: - json.dump( - { - "uploaded_file": str(input[0]), - "upload_time_secs": end_perf - start_perf, - "start_timestamp": start_timestamp, - "end_timestamp": end_timestamp, - }, - fh, - indent=0, - ) + # Success, create the sentinel file, which also contains a manifest for the + # uploaded file. + # Because we may re-upload for the same day on another occasion. + # create a dated sentinel file for just this upload, and then + # copy it to the "latest" sentinel file + # to tell snakemake that this upload is complete. The latter will get overwritten + # on each re-upload, but the dated ones will be kept. + overall_sentinel = Path(output.overall_uploaded_sentinel) + new_suffix = f".{file_safe_timestamp}{overall_sentinel.suffix}" + dated_sentinel_file = overall_sentinel.with_suffix(new_suffix) + write_ftps_sentinel( + { + "upload_time_secs": perf_time, + "start_timestamp": start_timestamp, + "end_timestamp": end_timestamp, + "local_tar_file": remote_tar_filename, + }, + dated_sentinel_file, + list(input.pseudonymised_parquets) + ) + shutil.copy(dated_sentinel_file, output.overall_uploaded_sentinel) diff --git a/src/pipeline/utils.py b/src/pipeline/utils.py index 7560b82..c16f0f1 100644 --- a/src/pipeline/utils.py +++ b/src/pipeline/utils.py @@ -1,5 +1,5 @@ import time -import telemetry + from datetime import datetime, timedelta, timezone from pathlib import Path import re @@ -9,12 +9,12 @@ from pseudon.hashing import do_hash from locations import ( WAVEFORM_PSEUDONYMISED_PARQUET, - WAVEFORM_FTPS_LOGS, HASH_LOOKUP_JSON, ORIGINAL_PARQUET_PATTERN, FILE_STEM_PATTERN_HASHED, CSV_PATTERN, make_file_name, + ALL_UPLOADED_JSON, ) @@ -67,9 +67,8 @@ def get_pseudonymised_parquet_path(self) -> Path: final_stem = make_file_name(FILE_STEM_PATTERN_HASHED, self._subs_dict) return WAVEFORM_PSEUDONYMISED_PARQUET / f"{final_stem}.parquet" - def get_ftps_uploaded_file(self) -> Path: - final_stem = make_file_name(FILE_STEM_PATTERN_HASHED, self._subs_dict) - return WAVEFORM_FTPS_LOGS / (final_stem + ".ftps.uploaded.json") + def get_ftps_uploaded_all_file(self) -> Path: + return Path(make_file_name(str(ALL_UPLOADED_JSON), self._subs_dict)) def get_daily_hash_lookup(self) -> Path: return Path(make_file_name(str(HASH_LOOKUP_JSON), self._subs_dict)) @@ -82,6 +81,12 @@ def get_file_age(file_path: Path) -> timedelta: return now_utc - file_time_utc +def timestamp_for_paths() -> str: + """A now timestamp that is safe for being in file paths on all OSes we are using.""" + now = datetime.now(timezone.utc) + return now.strftime("%Y-%m-%dT%H%M%SZ") + + def determine_eventual_outputs( csv_wait_time: timedelta, process_only_yesterday: bool, process_datestring: str ): @@ -90,7 +95,7 @@ def determine_eventual_outputs( :param process_only_yesterday: if false we process all dates, true only from yesterday :param process_datestring: a regular expression to match datestrings. Has no effect if process_only_yesterday is true - :returns: A list of InputCsvFile and a dictionary containing the hash and csn values. + :returns: A list of InputCsvFile and a dictionary containing the hashed csn -> csn mappings. """ # Discover all CSVs using the basic file name pattern before = time.perf_counter() @@ -132,7 +137,3 @@ def determine_eventual_outputs( f"Calculated output files using newness threshold {csv_wait_time} in {after - before} seconds" ) return _all_outputs, _hash_to_csn - - -def report_ftp_upload(): - telemetry.ftps_uploaded.add(1) diff --git a/src/telemetry.py b/src/telemetry.py index f0e5821..95492ab 100644 --- a/src/telemetry.py +++ b/src/telemetry.py @@ -49,7 +49,13 @@ ) ftps_uploaded = metrics.get_meter(INSTRUMENTATION_SCOPE).create_counter( - "waveform.ftps.uploaded", + "waveform.ftps.uploaded.count", unit="{file}", - description="Waveform ftps parquet file uploaded", + description="Waveform ftps parquet file upload status", +) + +ftps_time_taken = metrics.get_meter(INSTRUMENTATION_SCOPE).create_histogram( + "waveform.ftps.uploaded.time_taken", + unit="s", + description="Time taken to upload a days' messages", ) diff --git a/tests/test_ftps.py b/tests/test_ftps.py index df0dc41..24ba2f8 100644 --- a/tests/test_ftps.py +++ b/tests/test_ftps.py @@ -45,11 +45,11 @@ def test_do_upload_input_paths( # file needs to exist path_to_try.parent.mkdir(parents=True, exist_ok=True) path_to_try.write_text("blah") - ftps.do_upload(path_to_try) + ftps.do_upload_multiple([path_to_try], "dontcare.tar") assert connect_mock.called ftp_mock.storbinary.assert_called_once() else: # Don't create upload file as it may be outside the pytest tmp_path. We expect things to fail before that point anyway with pytest.raises(ValueError, match="must be under"): - ftps.do_upload(path_to_try) + ftps.do_upload_multiple([path_to_try], "dontcare.tar") connect_mock.assert_not_called()