Fix multiple FTPS upload issues

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`

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}"\

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,
+        )

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]):

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)