Sk/queries

.dockerignore

@@ -0,0 +1,6 @@
+**/.venv
+**/*.venv
+.idea
+.git
+**/.mypy_cache
+**/.pytest_tmp

.github/workflows/pytest.yml

@@ -54,6 +54,8 @@ jobs:
           cp config.EXAMPLE/hasher.env.EXAMPLE ../config/hasher.env
           cp config.EXAMPLE/controller.env.EXAMPLE ../config/controller.env
           cp config.EXAMPLE/monitoring.env.EXAMPLE ../config/monitoring.env
+          cp config.EXAMPLE/janitoring.env.EXAMPLE ../config/janitoring.env
+          # (lgtm is not involved in this test so its config file is not copied)
           {
             echo ""
             echo "AZURE_CLIENT_ID=${AZURE_CLIENT_ID}"
@@ -67,7 +69,7 @@ jobs:
 
       - name: Run the tests
         working-directory: waveform-controller
-        run: uv run pytest tests --cov=src --cov-report=term-missing --cov-report=xml
+        run: uv run pytest --cov --cov-report=term-missing --cov-report=xml
 
       - name: Upload coverage reports to Codecov
         uses: codecov/codecov-action@v5

.gitignore

@@ -11,6 +11,7 @@ wheels/
 
 # IDEs
 .idea/
+.vscode/
 
 # settings files (should not be in the source tree anyway, but just in case)
 *.env

config.EXAMPLE/controller.env.EXAMPLE

@@ -13,6 +13,7 @@ RABBITMQ_PASSWORD="my_pw"
 RABBITMQ_HOST="localhost"
 RABBITMQ_PORT=5672
 RABBITMQ_QUEUE="waveform"
+SQL_PATH="./src/sql/"
 # OpenTelemetry OTLP/HTTP endpoint of the LGTM collector.
 OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318"
 OTEL_SERVICE_NAME=waveform-controller

config.EXAMPLE/exporter.env.EXAMPLE

@@ -33,6 +33,33 @@ ONLY_USE_CSV_FROM_YESTERDAY=TRUE
 # expression to match multiple date
 PROCESS_CSV_FROM_DATE=
 
+# We query Caboodle to get electronic healthcare record date per patient per day
+CABOODLE_DBNAME="fakecab"
+CABOODLE_USERNAME="inform_user"
+CABOODLE_PASSWORD="inform"
+CABOODLE_HOST="localhost"
+CABOODLE_PORT="5433"
+CABOODLE_CONNECT_TIMEOUT="10"  # in seconds
+CABOODLE_QUERY_TIMEOUT="3000"  # in milliseconds
+
+# To avoid having to deploy a fake caboodle for testing we have 
+# a testing flag for Caboodle. If set TRUE caboodle connection will
+# fail silently and ehr file will be created with fake data
+CABOODLE_TESTING="FALSE"
+
+# The following is duplicated from controller.env
+# the exporter needs access to uds
+UDS_DBNAME="fakeuds"
+UDS_USERNAME="inform_user"
+UDS_PASSWORD="inform"
+UDS_HOST="172.17.0.1"
+UDS_PORT="5433"
+UDS_CONNECT_TIMEOUT="10"
+UDS_QUERY_TIMEOUT="3000"
+SCHEMA_NAME="schemaname"
+
+SQL_PATH="/app/src/sql/"
+
 # OpenTelemetry OTLP/HTTP endpoint of the LGTM collector.
 OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318"
 OTEL_SERVICE_NAME=waveform-exporter

config.EXAMPLE/janitoring.env.EXAMPLE

@@ -0,0 +1,13 @@
+# This is an EXAMPLE file, do not put real secrets in here.
+# Copy it to ../config/janitoring.env and then DELETE THIS COMMENT.
+# When does the janitoring job run
+JANITORING_CRON_SCHEDULE="*/15 * * * *"
+# OpenTelemetry OTLP/HTTP endpoint of the LGTM collector.
+OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318"
+OTEL_SERVICE_NAME=waveform-janitoring
+
+# Blank means do not delete
+HL7_BZ2_ARCHIVE_RETENTION_DAYS=
+ORIGINAL_CSV_RETENTION_DAYS=30
+ORIGINAL_PARQUET_RETENTION_DAYS=30
+PSEUDONYMISED_RETENTION_DAYS=30

docker-compose.yml

@@ -70,6 +70,7 @@ services:
     build:
       context: .
       dockerfile: monitoring/Dockerfile
+      target: supercronic_monitoring
       args:
         <<: *proxy-common
     env_file:
@@ -79,3 +80,17 @@ services:
       - ../../waveform-saved-messages:/waveform-saved-messages:ro
       - ../waveform-export:/waveform-export:ro
     restart: unless-stopped
+  waveform-janitoring:
+    build:
+      context: .
+      dockerfile: monitoring/Dockerfile
+      target: supercronic_janitoring
+      args:
+        <<: *proxy-common
+    env_file:
+      - ../config/janitoring.env
+    volumes:
+      # This will need to be rw if we ever enable deletion for real
+      - ../../waveform-saved-messages:/waveform-saved-messages:ro
+      - ../waveform-export:/waveform-export:ro
+    restart: unless-stopped

docs/deployment.md

@@ -35,6 +35,7 @@ sledgehammer approach which is rather similar to
 * Delete all Emap tables in `star_dev` as per Emap deployment instructions.
 * Waveform: `docker compose down` to bring everything down
 * git pull and rebuild containers for the two repos.
+* **REMEMBER: that the files from waveform-private-queries repository need to be copied to the waveform-controller/src/sql directory.**
 * Change config if necessary
 * Bring it all up again
 

docs/janitoring.md

@@ -0,0 +1,42 @@
+# File tidy up (waveform-janitoring)
+
+## Suggested retention times
+
+| Data                               | Size | Replaceability | Comment | Suggested deletion policy |
+|--------|--------|--------|--------|--------|
+| compressed HL7            | Large | Cannot be replaced |  | Never delete unless under duress. If disk gets low, try to find more disk, or delete old stuff first if we have confidence that our uploaded data will not need to be re-processed. Archives are indexed by time and bed, but ALL variables are mixed together, so targeted deletion is only possible by time and bed.  |
+| original CSV                    | Large but see #15  | Can be reprocessed from HL7 | can be useful for debugging | Delete after 30 days or if disk gets low; consider targeted deletion (see #79) |
+| original parquet               | Small | Can be reprocessed from CSV | parquets are generated in a single step from CSV | Delete after 30 days or if disk gets low; consider targeted deletion (see #79) |
+| pseduonymised parquet | Small | Can be reprocessed from CSV | parquets are generated in a single step from CSV | Delete after 30 days or if disk gets low; consider targeted deletion (see #79) |
+| pre-upload tar files         | Small | Is an intermediate file for upload | Not needed except for (short term) debugging | Delete after 30 days or if disk gets low; consider deleting immediately after upload | 
+
+Retention times for file types that are only useful for debugging could be shortened when we have more confidence in the pipeline.
+
+## Design decisions
+
+In the first iteration of this feature, retention times are evaluated against
+file modification times. Ie. processing time.
+
+It might, in future, be reasonable to also take observation time into account
+(ie. using the dirname "2024-10-01" to determine when the data relates to).
+
+For live data, these time stamps will be very similar so it doesn't matter which we use.
+
+But if we reprocess some data from stored HL7 archives,
+the modified times could be quite new vs the observation times.
+If we only use modification times, a large amount of data could be produced quite quickly
+that won't get cleared up, possibly leading to a full disk.
+If we only use observation times and the data is sufficiently old, we could see CSV files being
+deleted before we have a chance to convert and upload them.
+Therefore it seems likely that to solve those problems we'd have to somehow take both into account.
+
+There shouldn't be a scenario where observation times are newer than modification times
+(bar synthetic data).
+
+Snakemake has the ability to mark files as temporary. They are immediately deleted after they are needed.
+However, we want to keep files for a certain time after snakemake has finished, so we can't use it.
+
+
+## Configuration of janitoring service
+
+See [janitoring config example file](../config.EXAMPLE/janitoring.env.EXAMPLE) for config variables.

monitoring/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.13-slim-bookworm@sha256:8092ae2ef67061f9db412458dbdce44dbf16748fb3cae5cdbd020f467a9712d0
+FROM python:3.13-slim-bookworm@sha256:8092ae2ef67061f9db412458dbdce44dbf16748fb3cae5cdbd020f467a9712d0 AS supercronic_base
 LABEL authors="Stephen Thompson, Jeremy Stein"
 
 RUN export DEBIAN_FRONTEND=noninteractive && \
@@ -29,12 +29,15 @@ COPY --from=ghcr.io/astral-sh/uv@sha256:e85be844203885286c60ffad8a858d48afb6c5a5
 ARG UVCACHE=/root/.cache/uv
 WORKDIR /app
 
-COPY monitoring/monitor.py monitoring/monitor.py.lock /app/
-RUN uv lock --check --script monitor.py
+COPY monitoring/* /app/
+RUN chmod +x /app/*.sh
 
+FROM supercronic_base AS supercronic_monitoring
+RUN uv lock --check --script monitor.py
 RUN --mount=type=cache,target=${UVCACHE} uv export --script monitor.py --locked | uv pip install --system -r -
+ENTRYPOINT ["/app/monitoring_entrypoint.sh"]
 
-COPY monitoring/entrypoint.sh /app/entrypoint.sh
-RUN chmod +x /app/entrypoint.sh
-
-ENTRYPOINT ["/app/entrypoint.sh"]
+FROM supercronic_base AS supercronic_janitoring
+RUN uv lock --check --script janitor.py
+RUN --mount=type=cache,target=${UVCACHE} uv export --script janitor.py --locked | uv pip install --system -r -
+ENTRYPOINT ["/app/janitoring_entrypoint.sh"]

monitoring/janitor.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+"""Scan saved HL7 messages and emit OpenTelemetry metrics.
+
+Run in this command in dev to update the lockfile: `uv lock --script monitoring/janitor.py`
+"""
+
+import logging
+import sys
+from datetime import timedelta, datetime, timezone
+from pathlib import Path
+from time import perf_counter
+from typing import Optional
+
+from opentelemetry import metrics
+
+import utils
+
+# /// script
+# requires-python = ">=3.13"
+# dependencies = [
+#   "opentelemetry-exporter-otlp-proto-http==1.42.0",
+# ]
+# ///
+
+INSTRUMENTATION_SCOPE = "waveform-janitoring.meter"
+SAVED_MESSAGES_DIR = Path("/waveform-saved-messages")
+WAVEFORM_EXPORT_DIR = Path("/waveform-export")
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(message)s",
+    stream=sys.stdout,
+)
+logger = logging.getLogger(__name__)
+
+
+def scan_waveform_exporter_files(meter, dry_run):
+    scan_time_hist = meter.create_histogram(
+        "waveform.janitoring.exporter.disk_cleanup_time",
+        unit="s",
+        description="Duration of cleanup in exporter directory",
+    )
+    start_time = perf_counter()
+    # Dirs that contain large files where we need to clean up.
+    # Missing/blank env means do not clean up at all.
+    big_top_level_dirs: dict[Path, Optional[float]] = {
+        WAVEFORM_EXPORT_DIR / "original-csv": utils.get_env(
+            "ORIGINAL_CSV_RETENTION_DAYS", as_type=float
+        ),
+        WAVEFORM_EXPORT_DIR / "original-parquet": utils.get_env(
+            "ORIGINAL_PARQUET_RETENTION_DAYS", as_type=float
+        ),
+        WAVEFORM_EXPORT_DIR / "pseudonymised": utils.get_env(
+            "PSEUDONYMISED_RETENTION_DAYS", as_type=float
+        ),
+        SAVED_MESSAGES_DIR: utils.get_env(
+            "HL7_BZ2_ARCHIVE_RETENTION_DAYS", as_type=float
+        ),
+    }
+    bytes_deleted_histo = meter.create_histogram(
+        "waveform.janitoring.deleted_bytes",
+        unit="By",
+        description="Bytes deleted by the janitoring process",
+    )
+    for tld_name, retention_days in big_top_level_dirs.items():
+        logger.info(f"Scanning {tld_name} for items older than {retention_days} days")
+        if not retention_days:
+            logger.info("Skipping %s due to empty/missing retention value", tld_name)
+            continue
+        tld = WAVEFORM_EXPORT_DIR / tld_name
+        tld_meter_name = tld_name.name.replace("-", "_")
+        byte_count = _delete_old_files(tld, retention_days, dry_run)
+        bytes_deleted_histo.record(
+            byte_count,
+            attributes={
+                "directory": tld_meter_name,
+                "dry_run": bool(dry_run),
+            },
+        )
+    time_taken = perf_counter() - start_time
+    scan_time_hist.record(time_taken)
+    logger.info("Scanned %s in %ss", WAVEFORM_EXPORT_DIR, time_taken)
+
+
+def _delete_old_files(tld: Path, retention_days: float, dry_run) -> int:
+    retention_threshold = datetime.now(timezone.utc) - timedelta(days=retention_days)
+    retention_threshold_timestamp = retention_threshold.timestamp()
+    deleted_byte_count = 0
+    for dn, _, files in tld.walk():
+        for f in files:
+            f_path = dn / f
+            if f_path.is_file():
+                stat = f_path.stat()
+                actual_mtime = stat.st_mtime
+                if actual_mtime < retention_threshold_timestamp:
+                    if not dry_run:
+                        f_path.unlink()
+                    logger.info(
+                        "%sDeleting file [%s bytes] %s",
+                        "[DRY RUN] " if dry_run else "",
+                        stat.st_size,
+                        f_path,
+                    )
+                    deleted_byte_count += stat.st_size
+    return deleted_byte_count
+
+
+def main(args) -> int:
+    service_name = utils.get_env("OTEL_SERVICE_NAME")
+    otlp_endpoint = utils.get_env("OTEL_EXPORTER_OTLP_ENDPOINT")
+
+    # setup
+    utils.setup_metrics(service_name, otlp_endpoint)
+    meter = metrics.get_meter(INSTRUMENTATION_SCOPE)
+
+    # things to clean up
+    scan_waveform_exporter_files(meter, args.dry_run)
+
+    # shutdown, flush data
+    provider = metrics.get_meter_provider()
+
+    # (blank endpoint can return a provider that doesn't contain force_flush)
+    if hasattr(provider, "force_flush"):
+        provider.force_flush(timeout_millis=15000)
+
+    return 0
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+
+    raise SystemExit(main(args))