Fix multiple FTPS upload issues

.github/workflows/linting.yml

@@ -24,6 +24,6 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
         with:
-          python-version: "3.11"
+          python-version: "3.13"
           cache: "pip"
       - uses: pre-commit/action@v3.0.1

README.md

@@ -141,22 +141,7 @@ Each row of the csv will contain
 
 `csn, mrn, units, samplingRate, observationTime, waveformData`
 
-### Perform a parquet conversion (including de-id)
-At the time of writing, the cron pipeline is not set up. This section shows
-how to perform an ad-hoc de-id.
-```
-docker compose run waveform-controller emap-csv-pseudon  --csv /waveform-export/original-csv/my_original_csv.csv
-```
-
-### Perform an export
-At the time of writing, the cron pipeline is not set up. This section shows
-how to perform an ad-hoc FTPS upload.
-
-Exported files must be under the WAVEFORM_PSEUDONYMISED_PARQUET directory.
-Files passed in must be given relative to this directory:
-```
-docker compose run --entrypoint "" waveform-exporter emap-send-ftps my_pseudonymised_file.parquet
-```
+See [debugging doc to investigate further](docs/debugging.md)
 
 ## Developing
 See [developing docs](docs/develop.md)

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

pyproject.toml

@@ -35,8 +35,6 @@ coverage = [
 
 [project.scripts]
 emap-extract-waveform = "controller:receiver"
-emap-csv-pseudon = "pseudon.pseudon:psudon_cli"
-emap-send-ftps = "exporter.ftps:do_upload_cli"
 
 [tool.pytest.ini_options]
 # Force temp dirs under the repo so Docker can mount them on macOS.

src/exporter/ftps.py

@@ -1,62 +1,143 @@
-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()
-    parser.add_argument(
-        "file_to_upload",
-        type=Path,
-        help="file to upload relative to pseudonymised folder",
+
+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
     )
-    args = parser.parse_args()
-    do_upload(args.file_to_upload)
+    attrs = {
+        "obs_date": wc_date,
+        # It's possible you'd run staging+prod instances
+        # that feed into the same collector, so differentiate here.
+        "deployment.environment.name": settings.INSTANCE_NAME,
+    }
+    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_tars.add(1, attributes=attrs)
+        telemetry.ftps_uploaded_parquets.add(len(file_list), attributes=attrs)
+        telemetry.ftps_time_taken.record(perf_time)
+    return perf_time
 
 
-def do_upload(abs_file_to_upload: Path):
+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]):