diff --git a/annotation/annotation_run_files.py b/annotation/annotation_run_files.py index 999acdf4f..5ae6f146f 100644 --- a/annotation/annotation_run_files.py +++ b/annotation/annotation_run_files.py @@ -14,6 +14,11 @@ from library.utils.file_utils import name_from_filename from snpdb.variants_to_vcf import VARIANT_GRID_INFO_DICT, write_contig_sorted_values_to_vcf_file +# Prefix of a run's scratch dir under settings.IMPORT_PROCESSING_DIR (where the bulk inserter writes the +# CSVs it SQL COPYs from). Here rather than on the inserter so AnnotationRun can name the dir it owns +# without importing the annotation pipeline. +ANNOTATION_RUN_IMPORT_PROCESSING_PREFIX = "annotation_run" + def get_annotated_filename(annotation_run, vcf_dump_filename) -> str: """ Path VEP writes its annotated VCF to for a given dump. Derived from the dump stem, which #1658 diff --git a/annotation/fake_annotation.py b/annotation/fake_annotation.py index 8cbe3165e..5b11ebb6b 100644 --- a/annotation/fake_annotation.py +++ b/annotation/fake_annotation.py @@ -52,8 +52,7 @@ def get_fake_annotation_settings_dict(columns_version: int) -> dict: - TEST_IMPORT_PROCESSING_DIR = os.path.join(settings.PRIVATE_DATA_ROOT, 'import_processing', - "test", str(uuid4())) + TEST_IMPORT_PROCESSING_DIR = os.path.join(settings.IMPORT_PROCESSING_DIR, "test", str(uuid4())) TEST_ANNOTATION = copy.deepcopy(settings.ANNOTATION) # phastCons/phyloP custom tracks: v1-v3 fixtures were generated without the bigwig data, so disable diff --git a/annotation/management/commands/gene_annotation.py b/annotation/management/commands/gene_annotation.py index 85178891a..605ed18c0 100644 --- a/annotation/management/commands/gene_annotation.py +++ b/annotation/management/commands/gene_annotation.py @@ -20,7 +20,10 @@ from annotation.models.models import SubVersionPartition from genes.gene_matching import ReleaseGeneMatcher from genes.models import Gene, GeneAnnotationRelease, GnomADGeneConstraint, ReleaseGeneSymbolGene -from library.django_utils.django_file_utils import get_import_processing_filename +from library.django_utils.django_file_utils import ( + get_import_processing_filename, + remove_import_processing_dir, +) from ontology.models import ( ONTOLOGY_RELATIONSHIP_MEDIUM_QUALITY_FILTER, GeneDiseaseClassification, @@ -546,6 +549,8 @@ def _write_records(self, gene_annotation_version: GeneAnnotationVersion, gene_an self.stdout.write(f"Inserting file '{csv_filename}' into partition {partition_table}\n") sql_copy_csv(csv_filename, partition_table, self.GENE_ANNOTATION_HEADER, delimiter=delimiter) self.stdout.write("Done!\n") + if settings.IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS: + remove_import_processing_dir(gene_annotation_version.pk, prefix='gene_annotation') def bad_gene_annotation(): diff --git a/annotation/management/commands/human_protein_atlas_import.py b/annotation/management/commands/human_protein_atlas_import.py index f9912778a..26573b034 100644 --- a/annotation/management/commands/human_protein_atlas_import.py +++ b/annotation/management/commands/human_protein_atlas_import.py @@ -6,12 +6,16 @@ import os import pandas as pd +from django.conf import settings from django.core.management.base import BaseCommand, CommandError from annotation.models import HumanProteinAtlasAnnotationVersion, HumanProteinAtlasTissueSample from genes.models import Gene, GeneSymbol from genes.models_enums import AnnotationConsortium -from library.django_utils.django_file_utils import get_import_processing_filename +from library.django_utils.django_file_utils import ( + get_import_processing_filename, + remove_import_processing_dir, +) from library.utils import file_sha256sum from upload.vcf.sql_copy_files import sql_copy_csv, write_sql_copy_csv @@ -116,3 +120,5 @@ def handle(self, *args, **options): 'value'] sql_copy_csv(csv_filename, partition_table, HPA_HEADER, delimiter=delimiter) logging.info("Done!") + if settings.IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS: + remove_import_processing_dir(version_id, prefix='human_protein_atlas') diff --git a/annotation/models/models.py b/annotation/models/models.py index 4746f3320..7345bd0d0 100644 --- a/annotation/models/models.py +++ b/annotation/models/models.py @@ -11,7 +11,6 @@ import logging import os import re -import shutil from collections import defaultdict from collections.abc import Callable, Iterable from contextlib import contextmanager @@ -37,6 +36,7 @@ from psqlextra.models import PostgresPartitionedModel from psqlextra.types import PostgresPartitioningMethod +from annotation.annotation_run_files import ANNOTATION_RUN_IMPORT_PROCESSING_PREFIX from annotation.external_search_terms import ( get_variant_pubmed_search_terms, get_variant_search_terms, @@ -89,6 +89,7 @@ ) from genes.models_enums import AnnotationConsortium from library.django_utils import object_is_referenced +from library.django_utils.django_file_utils import remove_import_processing_dir from library.django_utils.data_archive_mixin import DataArchiveMixin from library.django_utils.django_partition import RelatedModelsPartitionModel from library.genomics import parse_gnomad_coord @@ -1420,10 +1421,8 @@ def reset_for_retry(self): # the upload_attempts>1 cleanup in import_vcf_annotations is skipped - so the leftover files trip # write_sql_copy_csv's "don't want to overwrite" guard, which is meant only for genuinely out-of-sync # dirs (moved dump / double launch). Done outside the transaction (filesystem op) and after the DB - # reset commits, so a rolled-back reset leaves the scratch dir intact. Prefix matches - # BulkVEPVCFAnnotationInserter.PREFIX. - import_processing_dir = os.path.join(settings.IMPORT_PROCESSING_DIR, f"annotation_run_{self.pk}") - shutil.rmtree(import_processing_dir, ignore_errors=True) + # reset commits, so a rolled-back reset leaves the scratch dir intact. + remove_import_processing_dir(self.pk, prefix=ANNOTATION_RUN_IMPORT_PROCESSING_PREFIX) def revert_external_to_local(self): """ #1568: return an external run to the normal local pipeline. Clears the external flag and dump diff --git a/annotation/vcf_files/bulk_vep_vcf_annotation_inserter.py b/annotation/vcf_files/bulk_vep_vcf_annotation_inserter.py index 1204e34f0..88d175724 100644 --- a/annotation/vcf_files/bulk_vep_vcf_annotation_inserter.py +++ b/annotation/vcf_files/bulk_vep_vcf_annotation_inserter.py @@ -9,7 +9,6 @@ import logging import operator import os -import shutil import time from collections import Counter, defaultdict from collections.abc import Iterable @@ -21,6 +20,7 @@ from django.conf import settings from annotation import vep_columns as vep_columns_registry +from annotation.annotation_run_files import ANNOTATION_RUN_IMPORT_PROCESSING_PREFIX from annotation.models.models import ( AnnotationRun, VariantAnnotation, @@ -45,8 +45,8 @@ from genes.models_enums import AnnotationConsortium from library.django_utils import get_model_fields from library.django_utils.django_file_utils import ( - get_import_processing_dir, get_import_processing_filename, + remove_import_processing_dir, ) from library.genomics import Range, overlap_fraction, parse_gnomad_coord from library.log_utils import log_traceback @@ -172,7 +172,7 @@ class BulkVEPVCFAnnotationInserter: VEP Fields are where they are copied are defined in ColumnVEPField """ - PREFIX = "annotation_run" + PREFIX = ANNOTATION_RUN_IMPORT_PROCESSING_PREFIX DB_FIXED_COLUMNS = [ "version_id", "annotation_run_id", @@ -958,10 +958,7 @@ def _gene_overlap_to_row_data(_header, annotations_list: list): pass def remove_processing_files(self): - import_processing_dir = get_import_processing_dir(self.annotation_run.pk, prefix=self.PREFIX) - logging.info("********* Deleting '%s' *******", import_processing_dir) - # ignore_errors so a missing dir (eg cleaned-up retry) doesn't blow up - we just want it gone - shutil.rmtree(import_processing_dir, ignore_errors=True) + remove_import_processing_dir(self.annotation_run.pk, prefix=self.PREFIX) @cached_property def gene_identifiers(self): diff --git a/claude/maps/commands.md b/claude/maps/commands.md index 7ece24490..51a8fd86c 100644 --- a/claude/maps/commands.md +++ b/claude/maps/commands.md @@ -53,6 +53,7 @@ Generated by `vg map commands` (do not edit; run `scripts/vg map commands` after | `import_dbnsfp_gene_annotation` | annotation | | annotation, genes | | `import_gene_annotation` | genes | | genes, snpdb | | `import_lab_info` | snpdb | | snpdb | +| `import_processing_cleanup` | upload | Remove import_processing scratch directories whose owner has finished with them (guessed) | annotation, snpdb, upload | | `import_sequencing_info` | seqauto | | seqauto, snpdb | | `import_transcript_sequence_fasta` | genes | | | | `import_vcf` | upload | | snpdb | diff --git a/claude/maps/signals.md b/claude/maps/signals.md index 5ad213d39..30473240d 100644 --- a/claude/maps/signals.md +++ b/claude/maps/signals.md @@ -56,6 +56,7 @@ Receivers of Django / third-party signals (post_save, pre_delete, m2m_changed, u | post_delete | VariantTag | `snpdb.management.commands.variant_tags:variant_tag_delete` | | post_delete | CohortGenotypeCollection | `snpdb.models.models_cohort:cohort_genotype_collection_post_delete_handler` | | post_delete | SubCohortVariantCollection | `snpdb.models.models_cohort:post_delete_sub_cohort_variant_collection` | +| post_delete | UploadPipeline | `upload.models.models:upload_pipeline_post_delete_handler` | | post_delete | UploadedPatientRecords | `upload.models.models_uploaded_files:uploaded_patient_records_post_delete_handler` | | post_save | ActiveSampleGeneList | `analysis.apps:handle_active_sample_gene_list_created` | | post_save | VariantTag | `analysis.apps:variant_tag_create` | diff --git a/claude/plans/928_import_processing_cleanup_plan.md b/claude/plans/928_import_processing_cleanup_plan.md deleted file mode 100644 index 6467d01a0..000000000 --- a/claude/plans/928_import_processing_cleanup_plan.md +++ /dev/null @@ -1,254 +0,0 @@ -# Clean up `import_processing` (#928) - -Written by Claude Fable 5 (claude-fable-5), 2026-08-31 - -## What the issue asks - -1. "A lot of processes don't clean up import_processing" – find out which, and why. -2. Consolidate `VCF_IMPORT_DELETE_TEMP_FILES_ON_SUCCESS` / `VARIANT_ANNOTATION_DELETE_TEMP_FILES_ON_SUCCESS` - into one setting. -3. (comment) `UploadPipelineFinishedTask` and `pipeline_success_task` look like duplicates – roll into one? - "One of the reasons for no cleanup is that UploadPipelineFinishedTask is setting status to SUCCESS while - pipeline success task only does anything if status = PROCESSING." - -Item 2 was done in 7f7a62d67 (both settings became `IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS`, -now `True` by default in `variantgrid/settings/components/default_settings.py:449`). Item 3 is the main -live bug and the comment's diagnosis is exactly right – details below. - -## Diagnosis - -### What is on disk (this dev box, `data/import_processing`, 5,276 entries, ~30 GB) - -| prefix | dirs | size | owner state | why it is still there | -|---|---|---|---|---| -| `pipeline_*` | 258 | 7.1 GB | 133 SUCCESS, 124 row deleted, 1 ERROR | §A (success path never runs), §B (nothing on delete) | -| `annotation_run_*` | 971 | 23 GB | all FINISHED, all pk ≤ 4472, none after 2026-07-06 | historical: setting was `not DEBUG` until f44fd2db3. Leak has stopped; needs a one-off sweep only | -| `clingen_allele_registry_` | 3,856 | 16 MB | – | all **empty**: §C | -| `liftover_*` / `manual_variants_*` / `classification_import_*` | 114 / 16 / 45 | ~2 MB | pipelines SUCCESS | §D (generated input VCF has no lifecycle hook) | -| `gene_annotation_*` | 3 | 9 MB | – | §E management commands leave their COPY CSV | -| `test/`, plus `pipeline_4919` etc. with no DB row | 26 + | small | – | §F tests write into the real dir | - -### A. `UploadPipelineFinishedTask` defeats `pipeline_success_task` (the comment's hunch, confirmed) - -`schedule_pipeline_stage_steps` (`upload/tasks/vcf/import_vcf_step_task.py:207-213`) builds the FINISH -stage as a **chain**: `[..., pipeline_success_task]`. For every -factory that inherits `AbstractVCFImportTaskFactory.get_finish_task_classes()` (`[UploadPipelineFinishedTask]`, -`abstract_vcf_import_task_factory.py:50`) the chain is therefore: - -1. `UploadPipelineFinishedTask.process_items` (`import_vcf_tasks.py:152`) – sets `status = SUCCESS`, saves. -2. `pipeline_success_task` (`import_vcf_step_task.py:233`) – guarded by `if status == PROCESSING`; it is - now SUCCESS, so it returns without calling `UploadPipeline.success()`. - -`UploadPipeline.success()` (`upload/models/models.py:226`) is the only place that (a) calls -`remove_processing_files()`, (b) records `processing_seconds_*`, (c) fires the `import_*_success` event. -None of that happens for those pipelines. - -DB proof (local): of 178 SUCCESS pipelines, 156 have `processing_seconds_wall_time IS NULL`. Every one of -those has `UploadPipelineFinishedTask` among its FINISH steps (Liftover, Manual Variant Entry, -Insert-variants-only, Variant Tags, TSO500 …). The 22 with timings are precisely the factories whose FINISH -list lacks it: ClinVar (`[ImportClinVarSuccessTask]`), Patient Records (no FINISH steps), and genotype VCF -(`settings.FINISH_IMPORT_VCF_STEP_TASKS_CLASSES = []`). Those three prove `pipeline_success_task` on its own -closes a pipeline correctly – FINISH is scheduled by `check_pipeline_stage` whether or not any -FINISH-dependent steps exist. - -So the answer to the comment is yes: `UploadPipelineFinishedTask` is redundant, and its presence is the -bug. The `status == PROCESSING` guard in `pipeline_success_task` must stay – FINISH can legitimately be -scheduled more than once (two DATA_INSERTION steps finishing together each call `check_pipeline_stage`; -`schedule_pipeline_stage_steps` de-duplicates the *steps* via `start_date` but appends -`pipeline_success_task` unconditionally), and the guard is what makes the second call a no-op. - -### B. Deleting an `UploadPipeline` row leaves its directory - -124 `pipeline_*` dirs have no row. There is no `post_delete` for `UploadPipeline` (only -`pre_delete_uploaded_vcf`). `annotation/signals/annotation_run_cleanup.py` already establishes the pattern -for AnnotationRun; UploadPipeline needs the same. - -### C. `get_import_processing_dir()` creates the directory on every call - -`library/django_utils/django_file_utils.py:8` – `mk_path` inside the getter. `ClinGenAlleleRegistryAPI.__init__` -(`snpdb/clingen_allele_api.py:73`) computes `api_failure_output_filename` eagerly for every instance, so each -of the ~7 `.instance()` call sites mints an empty `clingen_allele_registry_` dir per call. The file is -only ever written in the `except` branch of `_get_or_post` (line 122). Same getter also means every -`remove_processing_files()` recreates the dir before deleting it (harmless but backwards). - -### D. Generated input VCFs - -`snpdb/liftover.py:95`, `annotation/manual_variant_entry.py:80`, `classification/classification_import.py:122` -write a VCF into their own `_` dir, point `FileUpload.path` at it and run a pipeline. Nothing -removes that dir. (Variant Tags and TSO500 fusions write theirs *inside* `pipeline_` via a -`pre_vcf_task`, so they are regenerated on retry and cleaned with the pipeline – the right shape, but -moving liftover/manual/classification onto that shape is more than this issue needs.) - -Trade-off to accept: once removed, "Retry import" on an already-successful liftover / manual entry / -classification import is unavailable. `view_upload_pipeline` already handles a missing file (warning -"File does not exist on disk, cannot reload", `allow_retry_import=False`), and ERROR pipelines keep their -input, which is the case retry exists for. - -### E. Management commands - -`annotation/management/commands/gene_annotation.py:539` and `human_protein_atlas_import.py:103` COPY from a -CSV under `gene_annotation_` / `human_protein_atlas_` and leave it. `genes/models.py:2231` (gene -coverage) cleans up but gates on `not settings.DEBUG` instead of the setting. - -### F. Tests - -`annotation/fake_annotation.py:42` puts fake-annotation scratch under the *real* -`PRIVATE_DATA_ROOT/import_processing/test/` and never removes it; tests that build an `UploadPipeline` -without `override_settings(IMPORT_PROCESSING_DIR=…)` write `pipeline_` straight into the real -dir (`pipeline_4919`, mtime today, no row in the dev DB). - -### Settings - -`IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS` (import_processing scratch) and -`ANNOTATION_DELETE_TEMP_FILES_ON_SUCCESS` (#1670, `ANNOTATION_VCF_DUMP_DIR` – VEP input/output, AnnotSV -dir) govern different trees with different retention reasoning (a failed VEP run keeps its dump for -investigation and retry-upload-only). Keep both; this plan uses the import_processing one everywhere under -`IMPORT_PROCESSING_DIR`. - -## Plan - -### 1. One success path for VCF pipelines - -- Delete `UploadPipelineFinishedTask` (`upload/tasks/vcf/import_vcf_tasks.py:152-158` and the - `register_task` at line 282). Remove the import in `abstract_vcf_import_task_factory.py`. -- `AbstractVCFImportTaskFactory.get_finish_task_classes()` returns `[]`. -- `LiftoverImportFactory.get_finish_task_classes()` (`import_task_factories.py:403`) returns - `[LiftoverCompleteTask]`. -- `pipeline_success_task` stays as the single closer; keep its `PROCESSING` guard. Add a short docstring - saying it is the only thing that takes a VCF pipeline out of PROCESSING and why the guard exists. - The comment already on `GeneLevel…get_finish_task_classes` (`import_task_factories.py:168`) remains true. -- `UploadStep` rows on existing deployments that still name - `upload.tasks.vcf.import_vcf_tasks.UploadPipelineFinishedTask` in `script` belong to finished pipelines - and are never re-launched (`schedule_pipeline_stage_steps` filters `start_date__isnull=True`), so no data - migration. A retry of an old pipeline re-creates its steps from the factory. - -### 2. Path helpers that only create on write - -`library/django_utils/django_file_utils.py`: - -```python -def import_processing_dir_path(pk, prefix='pipeline') -> str: # pure – no mkdir -def get_import_processing_dir(pk, prefix='pipeline') -> str: # path + mk_path (existing callers) -def get_import_processing_filename(pk, base_filename, prefix='pipeline') -> str # unchanged -def remove_import_processing_dir(pk, prefix='pipeline'): # rmtree(ignore_errors=True) of the pure path, logs -``` - -Use `remove_import_processing_dir` from: -- `UploadPipeline.remove_processing_files()` (`upload/models/models.py:210`) – also drops the bare - `rmtree` that raises if the dir is gone. -- `BulkVEPVCFAnnotationInserter.remove_processing_files()` (`bulk_vep_vcf_annotation_inserter.py:952`). -- `AnnotationRun.reset_for_retry` (`annotation/models/models.py:1376`) – replace the hard-coded - `f"annotation_run_{self.pk}"` with the helper. Put the prefix constant - `ANNOTATION_RUN_IMPORT_PROCESSING_PREFIX = "annotation_run"` in `annotation/annotation_run_files.py` - (imports only library/snpdb, so both models and the bulk inserter can import it) and have - `BulkVEPVCFAnnotationInserter.PREFIX` read it. - -### 3. Pipeline directory lifecycle - -`upload/models/models.py`: -- `UploadPipeline.success()` keeps calling `remove_processing_files()` under the setting (now reachable - thanks to §1) and additionally calls a new `remove_generated_input_file()` (see §4). -- New `post_delete` receiver for `UploadPipeline` next to `pre_delete_uploaded_vcf` (`models.py:528`): - `remove_processing_files()` + `remove_generated_input_file()`, unconditional – with the row gone nothing - can name the directory (same reasoning as `annotation_run_post_delete_handler`). Registering a receiver - also disables Django's fast-delete path for queryset deletes, which is what makes it fire for - `FileUpload`/`VCF` cascades. -- ERROR pipelines keep everything on disk (matches #1670's decision for AnnotationRun); retry already - wipes the pipeline dir before re-running. - -### 4. Generated input VCFs (liftover / manual variants / classification import) - -- `FileUpload.is_import_processing_scratch` property: `path` is set, `import_source != WEB_UPLOAD`, and - `os.path.realpath(path)` is under `settings.IMPORT_PROCESSING_DIR`. -- `UploadPipeline.remove_generated_input_file()`: when the file is scratch **and** it is outside the - pipeline's own dir, `rmtree(dirname(path), ignore_errors=True)`. (Files inside `pipeline_` – variant - tags, TSO500 – are already covered by `remove_processing_files`.) -- Called from `success()` (gated on the setting) and from the `post_delete` receiver (unconditional). - `retry_upload_pipeline` (`upload/uploaded_file_type.py:98`) keeps calling only `remove_processing_files()` - so a retry still has its input. -- `LiftoverRun.source_vcf` stays as a record of what was written, like AnnotationRun's filename fields. - -### 5. ClinGen failure dump - -`snpdb/clingen_allele_api.py`: drop the constructor's eager `get_import_processing_filename`. Keep the -`api_failure_output_filename` kwarg for callers/tests that pass one; when it is `None`, compute the path -inside the `except` branch of `_get_or_post` immediately before writing, as -`get_import_processing_filename("failures", f"{uuid4()}.json", prefix="clingen_allele_registry")` – one -shared `clingen_allele_registry_failures/` dir, created only when a failure is actually dumped. Check -`snpdb/tests/utils/mock_clingen_api.py` and `snpdb/tests/test_clingen_allele.py` for anything relying on -the attribute being pre-set. - -### 6. Management commands and gene coverage - -- `gene_annotation.py::_write_records` and `human_protein_atlas_import.py`: after `sql_copy_csv`, if - `settings.IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS`, `remove_import_processing_dir(pk, prefix)`. -- `genes/models.py:2231`: gate on `settings.IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS` instead of - `not settings.DEBUG`, and go through the pure-path helper (`gene_coverage/gene_coverage_collection_` is - a nested layout – keep it, just replace the inline `rmtree`). -- `snpdb/tasks/somalier_tasks.py` already cleans up; leave as is. - -### 7. Tests write to a throwaway directory - -`variantgrid/settings/components/default_settings.py`, inside an existing `if UNIT_TEST:` block: - -```python -IMPORT_PROCESSING_DIR = os.path.join(tempfile.gettempdir(), "variantgrid_unit_test", "import_processing") -``` - -and `VariantGridTestRunner.teardown_test_environment` removes it. `annotation/fake_annotation.py:42` then -builds its per-test dir as `os.path.join(settings.IMPORT_PROCESSING_DIR, "test", str(uuid4()))` so it lands -in the same tree. Existing `override_settings(IMPORT_PROCESSING_DIR=tempdir)` in individual tests keep -working. - -### 8. Sweep for existing deployments - -New management command `upload/management/commands/import_processing_cleanup.py` (`--dry-run` flag, -prints per-prefix counts and bytes). It walks `settings.IMPORT_PROCESSING_DIR` top level and removes: - -| entry | remove when | -|---|---| -| `pipeline_` | no `UploadPipeline` row, or `status == SUCCESS` | -| `annotation_run_` | no `AnnotationRun` row, or `get_status() == FINISHED` | -| `clingen_allele_registry_` | directory is empty | -| `liftover_` | `UploadedLiftover(liftover_id=pk)` missing, or its pipeline SUCCESS/missing | -| `manual_variants_` | `UploadedManualVariantEntryCollection(collection_id=pk)` – same rule | -| `classification_import_` | `UploadedClassificationImport(classification_import_id=pk)` – same rule | -| `somalier_vcf_extract_` / `somalier_relate_` | owning row missing or status SUCCESS | -| `gene_annotation_` / `human_protein_atlas_` | owning version row exists | -| `test/` | always | - -Anything else (unknown prefix, ERROR/PROCESSING owners) is listed and left alone. Add -`upload/migrations/00XX_one_off_import_processing_cleanup.py` with -`ManualOperation.task_id_manage(["import_processing_cleanup"])` and a `test=` that returns True only when -`IMPORT_PROCESSING_DIR` exists and is non-empty, so the upgrade script surfaces it where there is something -to reclaim. The command is worth keeping (developers running with the setting off can run it by hand). - -### 9. Tests - -- `upload/tests/test_pipeline_success.py`: pipeline in PROCESSING with a FINISH-dependent step; run - `pipeline_success_task` → status SUCCESS, `processing_seconds_wall_time` set, pipeline dir removed with the - setting True and kept with it False; a second call is a no-op. -- `post_delete` on `UploadPipeline` removes the pipeline dir and a scratch input dir; a WEB_UPLOAD - `FileUpload` (path under upload storage) is left alone. -- `retry_upload_pipeline` removes the pipeline dir and keeps the generated input. -- `import_processing_dir_path` creates nothing; `ClinGenAlleleRegistryAPI()` creates nothing. -- `import_processing_cleanup --dry-run` / real run against a tempdir seeded with one entry per row of the - §8 table (owner present-and-successful, owner present-and-error, owner missing). - -Audit at the end per CLAUDE.md – drop any that only restate framework behaviour. - -## Files - -- `library/django_utils/django_file_utils.py` -- `upload/models/models.py` (success, remove_*, post_delete receiver, `FileUpload.is_import_processing_scratch`) -- `upload/tasks/vcf/import_vcf_tasks.py`, `upload/tasks/vcf/import_vcf_step_task.py` -- `upload/import_task_factories/abstract_vcf_import_task_factory.py`, `import_task_factories.py` -- `upload/management/commands/import_processing_cleanup.py` + `upload/migrations/00XX_…py` -- `annotation/annotation_run_files.py`, `annotation/models/models.py`, - `annotation/vcf_files/bulk_vep_vcf_annotation_inserter.py` -- `annotation/management/commands/gene_annotation.py`, `human_protein_atlas_import.py` -- `annotation/fake_annotation.py`, `variantgrid/test_runner.py`, - `variantgrid/settings/components/default_settings.py` -- `genes/models.py`, `snpdb/clingen_allele_api.py` -- `variantgrid/templates/default_templates/changelog.html` – entry for #928 -- tests as in §9 diff --git a/claude/research/upload.md b/claude/research/upload.md index bc4985fba..6447fc73d 100644 --- a/claude/research/upload.md +++ b/claude/research/upload.md @@ -189,6 +189,13 @@ and `upload/signals/signals.py:vcf_import_success_signal` for everyone else, and `pipeline_success_task` totals wall and CPU seconds from the step rows and `UploadPipeline.success` deletes the processing directory when `IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS` is set. +`pipeline_success_task` is the only thing that takes a pipeline out of PROCESSING, and its `status == PROCESSING` guard +is load-bearing: FINISH can be scheduled more than once, and the guard is what makes the second run a no-op. That makes +it fragile in one direction - a FINISH step that sets SUCCESS itself swallows the real close, taking the timings, the +success Event and the cleanup with it. #928 removed the step that did (`UploadPipelineFinishedTask`), so a FINISH task +class must leave the status alone. Deleting the pipeline row cleans up too: `upload/models/models.py:upload_pipeline_post_delete_handler` +removes the processing directory and, for a generated input VCF, the directory holding it. + ### Failure and retry Failure is one-way and top-down: `UploadStep.error_exception` stores the traceback and calls diff --git a/genes/models/models_gene_coverage.py b/genes/models/models_gene_coverage.py index cb213204a..789b58931 100644 --- a/genes/models/models_gene_coverage.py +++ b/genes/models/models_gene_coverage.py @@ -1,6 +1,5 @@ import logging import os -import shutil from collections import defaultdict from django.conf import settings @@ -19,9 +18,12 @@ from genes.models.models_gene import GeneSymbol, Transcript, TranscriptVersion from genes.models_enums import AnnotationConsortium from library.django_utils.data_archive_mixin import DataArchiveMixin +from library.django_utils.django_file_utils import ( + get_import_processing_dir, + remove_import_processing_dir, +) from library.django_utils.django_partition import RelatedModelsPartitionModel from library.log_utils import log_traceback -from library.utils.file_utils import mk_path from snpdb.archive import DataArchivedError from snpdb.models import DataState from snpdb.models.models_genome import GenomeBuild @@ -32,6 +34,10 @@ write_sql_copy_csv, ) +# Nested a level down from the other import_processing prefixes - a sequencing run loads coverage for +# every sample, so these are kept out of the top level the cleanup command sweeps +GENE_COVERAGE_IMPORT_PROCESSING_PREFIX = os.path.join("gene_coverage", "gene_coverage_collection") + class CanonicalTranscriptCollection(TimeStampedModel): description = models.TextField(blank=True) @@ -179,9 +185,7 @@ def get_examples(iterable): f"Sample transcripts from canonical transcript collection: {sample_canonical_transcripts}. ") raise ValueError(message) - processing_dir = os.path.join(settings.IMPORT_PROCESSING_DIR, "gene_coverage", - f"gene_coverage_collection_{self.pk}") - mk_path(processing_dir) + processing_dir = get_import_processing_dir(self.pk, GENE_COVERAGE_IMPORT_PROCESSING_PREFIX) if gene_coverage_tuples: csv_filename = os.path.join(processing_dir, f"gene_coverage_{self.pk}.csv") write_sql_copy_csv(gene_coverage_tuples, csv_filename) @@ -197,8 +201,8 @@ def get_examples(iterable): else: logging.warning("GeneCoverage had no canonical transcripts") - if not settings.DEBUG: - shutil.rmtree(processing_dir) + if settings.IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS: + remove_import_processing_dir(self.pk, GENE_COVERAGE_IMPORT_PROCESSING_PREFIX) logging.info("%d missing genes, %d missing transcripts", missing_genes, missing_transcripts) return warnings diff --git a/library/django_utils/django_file_utils.py b/library/django_utils/django_file_utils.py index c1a0388f6..802a0046b 100644 --- a/library/django_utils/django_file_utils.py +++ b/library/django_utils/django_file_utils.py @@ -1,22 +1,40 @@ """ Where an import pipeline's scratch files go: get_import_processing_dir / get_import_processing_filename under settings.PRIVATE_DATA_ROOT, keyed on the pipeline pk. + +get_import_processing_dir creates the directory, so it is for writers only. Anything that just wants to +name a directory - to remove it, or to test whether a path is under it - uses import_processing_dir_path, +otherwise merely asking where a thing would be leaves an empty directory behind (#928). """ +import logging import os +import shutil from django.conf import settings from library.utils.file_utils import mk_path -def get_import_processing_dir(pk, prefix='pipeline'): - output_dir = f"{prefix}_{pk}" - upd = os.path.join(settings.IMPORT_PROCESSING_DIR, output_dir) +def import_processing_dir_path(pk, prefix='pipeline') -> str: + """ Where this pk's scratch dir is (or would be) - creates nothing """ + return os.path.join(settings.IMPORT_PROCESSING_DIR, f"{prefix}_{pk}") + + +def get_import_processing_dir(pk, prefix='pipeline') -> str: + upd = import_processing_dir_path(pk, prefix) mk_path(upd) return upd -def get_import_processing_filename(pk, base_filename, prefix='pipeline'): +def get_import_processing_filename(pk, base_filename, prefix='pipeline') -> str: processing_dir = get_import_processing_dir(pk, prefix) filename = os.path.join(processing_dir, base_filename) return filename + + +def remove_import_processing_dir(pk, prefix='pipeline'): + """ Best effort - a dir that's already gone (eg a retry cleaned it up) is not an error """ + import_processing_dir = import_processing_dir_path(pk, prefix) + if os.path.exists(import_processing_dir): + logging.info("Removing import processing dir: '%s'", import_processing_dir) + shutil.rmtree(import_processing_dir, ignore_errors=True) diff --git a/snpdb/clingen_allele_api.py b/snpdb/clingen_allele_api.py index 16deb4e79..f4a3b3f30 100644 --- a/snpdb/clingen_allele_api.py +++ b/snpdb/clingen_allele_api.py @@ -70,12 +70,16 @@ def instance(cls, **kwargs) -> 'ClinGenAlleleRegistryAPI': def __init__(self, api_failure_output_filename=None): self.login = settings.CLINGEN_ALLELE_REGISTRY_LOGIN self.password = settings.CLINGEN_ALLELE_REGISTRY_PASSWORD - if api_failure_output_filename is None: - api_failure_output_filename = get_import_processing_filename(uuid.uuid4(), - "api_failure_output_filename.json", - prefix="clingen_allele_registry") + # Left None unless a caller names one - the path is only worked out when there's a failure to dump. + # Computing it here minted an empty import_processing dir per instance (#928) self.api_failure_output_filename = api_failure_output_filename + def _get_api_failure_output_filename(self) -> str: + if self.api_failure_output_filename is None: + self.api_failure_output_filename = get_import_processing_filename( + "failures", f"{uuid.uuid4()}.json", prefix="clingen_allele_registry") + return self.api_failure_output_filename + @staticmethod def check_api_response(api_response): """ Throws ClinGenAlleleAPIException if 'errorType' set """ @@ -113,19 +117,17 @@ def _put(self, url, data, chunk_size=None): self._check_response(response) return response.json() except Exception as e: - if self.api_failure_output_filename: - api_failure = { - "request": request, - "timeout": timeout, - "data": data, - } - with open(self.api_failure_output_filename, "w") as f: - json.dump(api_failure, f) - - msg = f"API call failed, debug info written to '{self.api_failure_output_filename}'" - raise ClinGenAllele.ClinGenAlleleRegistryException(msg) from e - else: - raise e + api_failure = { + "request": request, + "timeout": timeout, + "data": data, + } + api_failure_output_filename = self._get_api_failure_output_filename() + with open(api_failure_output_filename, "w") as f: + json.dump(api_failure, f) + + msg = f"API call failed, debug info written to '{api_failure_output_filename}'" + raise ClinGenAllele.ClinGenAlleleRegistryException(msg) from e @classmethod def get_code(cls, code): diff --git a/upload/CLAUDE.md b/upload/CLAUDE.md index 78dd7ec17..e542aec77 100644 --- a/upload/CLAUDE.md +++ b/upload/CLAUDE.md @@ -68,6 +68,13 @@ Gotchas: schedule_pipeline_stage_steps; everything else lands on db_workers. - settings.UPLOAD_ENABLED=False makes InsertUnknownVariantsTask raise; IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS wipes the processing dir on success, so inspect a failed pipeline's files before retrying it. +- upload/tasks/vcf/import_vcf_step_task.py:pipeline_success_task is the only thing that closes a VCF pipeline, and it is + guarded on status == PROCESSING (FINISH can be scheduled twice). A FINISH task that sets SUCCESS itself therefore + silently swallows the timings, the success Event and the file cleanup - #928 was exactly that. Leave the status alone + in get_finish_task_classes tasks. +- library/django_utils/django_file_utils.py:get_import_processing_dir creates the directory; use + import_processing_dir_path when you only want to name one, and remove_import_processing_dir to remove it. + manage.py import_processing_cleanup --dry-run reports what is reclaimable under settings.IMPORT_PROCESSING_DIR. Tests: - Whole pipeline in-process: create a FileUpload, then process_uploaded_file(file_upload, run_async=False) under CELERY_TASK_ALWAYS_EAGER (library/django_utils/unittest_utils.py:URLTestCase sets it) — diff --git a/upload/import_task_factories/abstract_vcf_import_task_factory.py b/upload/import_task_factories/abstract_vcf_import_task_factory.py index 74737d1a6..3ed081a5f 100644 --- a/upload/import_task_factories/abstract_vcf_import_task_factory.py +++ b/upload/import_task_factories/abstract_vcf_import_task_factory.py @@ -14,7 +14,6 @@ DoNothingVCFTask, PreprocessVCFTask, ScheduleMultiFileOutputTasksTask, - UploadPipelineFinishedTask, ) @@ -48,7 +47,9 @@ def get_post_data_insertion_classes(self): return [] def get_finish_task_classes(self): - return [UploadPipelineFinishedTask] + """ Steps run at the end of the FINISH chain, before pipeline_success_task closes the pipeline. + Anything here must leave the status PROCESSING - @see pipeline_success_task """ + return [] def get_pre_vcf_task(self, upload_pipeline): """ Run before loading initial VCF (use for e.g. retrieving/making it) """ diff --git a/upload/import_task_factories/import_task_factories.py b/upload/import_task_factories/import_task_factories.py index eb443c28d..59ef70f0b 100644 --- a/upload/import_task_factories/import_task_factories.py +++ b/upload/import_task_factories/import_task_factories.py @@ -401,8 +401,7 @@ def get_post_data_insertion_classes(self): return [VCFCheckAnnotationTask] def get_finish_task_classes(self): - task_classes = super().get_finish_task_classes() - return [LiftoverCompleteTask] + task_classes + return [LiftoverCompleteTask] class VariantTagsImportTaskFactory(VCFInsertVariantsOnlyImportFactory): diff --git a/upload/management/commands/import_processing_cleanup.py b/upload/management/commands/import_processing_cleanup.py new file mode 100644 index 000000000..b27de5e12 --- /dev/null +++ b/upload/management/commands/import_processing_cleanup.py @@ -0,0 +1,222 @@ +""" +Sweeps settings.IMPORT_PROCESSING_DIR of scratch directories whose owner has finished with them (#928). + +Everything under that directory is disposable working space written by an import, but until #928 several +paths never removed theirs, so an existing deployment has years of them. Each owner now cleans up as it +finishes (UploadPipeline.success and its post_delete, the annotation run reset, the SQL COPY commands), +so this is mostly a one-off catch-up - registered as a ManualOperation in +upload/migrations/0042_one_off_import_processing_cleanup.py - but it is worth keeping: a deployment +running with IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS off still accumulates them. + +An entry is only removed when its owner says it is done with it. Anything still owned (a PROCESSING +pipeline, an ERROR annotation run kept for investigation), anything modified within --min-age-days, and +anything whose name this does not recognise is reported and left alone. +""" +import os +import shutil +import time + +from django.conf import settings +from django.core.management.base import BaseCommand + +from annotation.models import ( + AnnotationRun, + GeneAnnotationVersion, + HumanProteinAtlasAnnotationVersion, +) +from annotation.models.models_enums import AnnotationStatus +from snpdb.models import SomalierVCFExtract +from snpdb.models.models_enums import ProcessingStatus +from upload.models import ( + UploadedClassificationImport, + UploadedLiftover, + UploadedManualVariantEntryCollection, + UploadPipeline, +) + +DAY_SECS = 24 * 60 * 60 + + +def _dir_size(path: str) -> int: + total = 0 + for dir_path, _dir_names, filenames in os.walk(path): + for filename in filenames: + try: + total += os.lstat(os.path.join(dir_path, filename)).st_size + except OSError: + pass + return total + + +def _pipeline_finished(file_upload_id=None, pk=None) -> bool: + """ A pipeline that's gone, or succeeded, has no further use for its files. ERROR/PROCESSING keeps + them - that is what 'Retry import' reloads from, and what a failure is investigated with """ + if pk is not None: + qs = UploadPipeline.objects.filter(pk=pk) + else: + qs = UploadPipeline.objects.filter(file_upload_id=file_upload_id) + status = qs.values_list("status", flat=True).first() + return status is None or status == ProcessingStatus.SUCCESS + + +def _upload_pipeline_finished(pk: int) -> bool: + """ pipeline_ - split VCF chunks, per-step logs and COPY CSVs of a VCF import """ + return _pipeline_finished(pk=pk) + + +def _annotation_run_finished(pk: int) -> bool: + """ annotation_run_ - the CSVs BulkVEPVCFAnnotationInserter SQL COPYs from """ + if annotation_run := AnnotationRun.objects.filter(pk=pk).first(): + return annotation_run.get_status() == AnnotationStatus.FINISHED + return True + + +def _generated_input_finished(upload_data) -> bool: + """ liftover_ / manual_variants_ / classification_import_ hold a VCF we generated as the + input to a pipeline, so they live as long as that pipeline wants to be able to re-read it """ + if upload_data is None: + return True + return _pipeline_finished(file_upload_id=upload_data.file_upload_id) + + +def _liftover_finished(pk: int) -> bool: + return _generated_input_finished(UploadedLiftover.objects.filter(liftover_id=pk).first()) + + +def _manual_variants_finished(pk: int) -> bool: + return _generated_input_finished(UploadedManualVariantEntryCollection.objects.filter(collection_id=pk).first()) + + +def _classification_import_finished(pk: int) -> bool: + return _generated_input_finished(UploadedClassificationImport.objects.filter(classification_import_id=pk).first()) + + +def _somalier_vcf_extract_finished(pk: int) -> bool: + status = SomalierVCFExtract.objects.filter(pk=pk).values_list("status", flat=True).first() + return status is None or status == ProcessingStatus.SUCCESS + + +def _somalier_relate_finished(_pk: int) -> bool: + """ snpdb.tasks.somalier_tasks._somalier_relate removes its own dir the moment somalier returns, so + one that survived belongs to a run that died. SomalierRelate is abstract (a table per subclass) + so the pk in the name doesn't identify a row anyway - the age guard is what keeps a live run safe """ + return True + + +def _gene_annotation_finished(pk: int) -> bool: + """ The COPY runs after the version row is created, so a row means the import reached the insert """ + return GeneAnnotationVersion.objects.filter(pk=pk).exists() + + +def _human_protein_atlas_finished(pk: int) -> bool: + return HumanProteinAtlasAnnotationVersion.objects.filter(pk=pk).exists() + + +# _ directory name -> is its owner finished with it? Longest prefix wins when matching +PK_OWNERS = { + "pipeline": _upload_pipeline_finished, + "annotation_run": _annotation_run_finished, + "liftover": _liftover_finished, + "manual_variants": _manual_variants_finished, + "classification_import": _classification_import_finished, + "somalier_vcf_extract": _somalier_vcf_extract_finished, + "somalier_relate": _somalier_relate_finished, + "gene_annotation": _gene_annotation_finished, + "human_protein_atlas": _human_protein_atlas_finished, +} + +CLINGEN_PREFIX = "clingen_allele_registry_" +UNIT_TEST_DIR_NAME = "test" +# genes.models.models_gene_coverage nests a dir per collection under here rather than at the top level - +# a sequencing run loads coverage for every sample, so there would be thousands of top level entries +GENE_COVERAGE_DIR_NAME = "gene_coverage" + + +class Command(BaseCommand): + help = "Remove import_processing scratch directories whose owner has finished with them" + + def add_arguments(self, parser): + parser.add_argument('--dry-run', action='store_true', + help="Report what would be removed, removing nothing") + parser.add_argument('--min-age-days', type=float, default=1, + help="Leave entries modified more recently than this alone (default: 1)") + + def handle(self, *args, **options): + dry_run = options["dry_run"] + cutoff = time.time() - options["min_age_days"] * DAY_SECS + import_processing_dir = settings.IMPORT_PROCESSING_DIR + if not os.path.isdir(import_processing_dir): + self.stdout.write(f"'{import_processing_dir}' does not exist - nothing to do\n") + return + + removed = {} + kept = {} + for name in sorted(os.listdir(import_processing_dir)): + path = os.path.join(import_processing_dir, name) + keep_reason = self._keep_reason(path, name, cutoff) + self._tally(kept if keep_reason else removed, path, name, keep_reason) + if keep_reason: + continue + if not dry_run: + if os.path.isdir(path): + shutil.rmtree(path, ignore_errors=True) + else: + os.remove(path) + + self._report("Would remove" if dry_run else "Removed", removed) + self._report("Kept", kept) + + def _keep_reason(self, path: str, name: str, cutoff: float) -> str: + """ Why this entry stays; empty string means it can go """ + try: + modified = os.lstat(path).st_mtime + except OSError: + return "unreadable" + if modified > cutoff: + return "modified recently" + + if name == UNIT_TEST_DIR_NAME: + return "" # Unit test scratch - @see annotation.fake_annotation + if name == GENE_COVERAGE_DIR_NAME: + return "gene coverage collections clean up their own dirs" + if name.startswith(CLINGEN_PREFIX): + # Empty ones were minted by ClinGenAlleleRegistryAPI merely being constructed (#928); + # a non-empty one holds an API failure dump someone may still want to read + if os.path.isdir(path) and not os.listdir(path): + return "" + return "holds ClinGen API failure dumps" + + prefix, pk = self._split_prefix_pk(name) + if prefix is None: + return "unrecognised" + if PK_OWNERS[prefix](pk): + return "" + return "owner not finished" + + @staticmethod + def _split_prefix_pk(name: str) -> tuple: + for prefix in sorted(PK_OWNERS, key=len, reverse=True): + if name.startswith(prefix + "_"): + pk = name[len(prefix) + 1:] + if pk.isdigit(): + return prefix, int(pk) + return None, None + + def _tally(self, counts: dict, path: str, name: str, keep_reason: str): + prefix, _pk = self._split_prefix_pk(name) + label = prefix or (CLINGEN_PREFIX.rstrip("_") if name.startswith(CLINGEN_PREFIX) else name) + if keep_reason: + label = f"{label} ({keep_reason})" + try: + size = _dir_size(path) if os.path.isdir(path) else os.lstat(path).st_size + except OSError: + size = 0 + num, total_size = counts.get(label, (0, 0)) + counts[label] = (num + 1, total_size + size) + + def _report(self, heading: str, counts: dict): + if not counts: + return + self.stdout.write(f"{heading}:\n") + for label, (num, size) in sorted(counts.items()): + self.stdout.write(f" {label}: {num} ({size / 1024 / 1024:.1f} MB)\n") diff --git a/upload/migrations/0042_one_off_import_processing_cleanup.py b/upload/migrations/0042_one_off_import_processing_cleanup.py new file mode 100644 index 000000000..e5702655d --- /dev/null +++ b/upload/migrations/0042_one_off_import_processing_cleanup.py @@ -0,0 +1,29 @@ +import os + +from django.conf import settings +from django.db import migrations + +from manual.operations.manual_operations import ManualOperation + + +def _has_import_processing_files(apps): # pylint: disable=unused-argument + """ #928: several paths never removed their import_processing scratch, so an existing deployment has + years of it. Only worth surfacing where there is something to reclaim """ + import_processing_dir = settings.IMPORT_PROCESSING_DIR + try: + return bool(os.listdir(import_processing_dir)) + except OSError: + return False + + +class Migration(migrations.Migration): + dependencies = [ + ("upload", "0041_alter_fileupload_file_type_and_more"), + ] + + operations = [ + ManualOperation(task_id=ManualOperation.task_id_manage(["import_processing_cleanup"]), + note="Reclaim import_processing scratch left behind by imports that never " + "cleaned up (#928). Run with --dry-run first to see what it would remove", + test=_has_import_processing_files), + ] diff --git a/upload/models/models.py b/upload/models/models.py index 3d8c1507d..e842a9b87 100644 --- a/upload/models/models.py +++ b/upload/models/models.py @@ -15,7 +15,7 @@ from django.db.models.aggregates import Max from django.db.models.deletion import CASCADE, SET_NULL from django.db.models.query import QuerySet -from django.db.models.signals import post_save, pre_delete +from django.db.models.signals import post_delete, post_save, pre_delete from django.dispatch.dispatcher import receiver from django.urls import reverse from django.utils import timezone @@ -26,7 +26,11 @@ from annotation.models.models_enums import VariantAnnotationPipelineType from eventlog.models import create_event from library.django_utils.django_file_system_storage import PrivateUploadStorage -from library.django_utils.django_file_utils import get_import_processing_dir +from library.django_utils.django_file_utils import ( + get_import_processing_dir, + import_processing_dir_path, + remove_import_processing_dir, +) from library.enums.log_level import LogLevel from library.log_utils import report_exc_info, report_message from library.utils import file_sha256sum @@ -81,6 +85,18 @@ def size(self) -> int: return self.file_field.size return os.stat(self.get_filename()).st_size + @property + def is_import_processing_scratch(self) -> bool: + """ True for a VCF we generated into IMPORT_PROCESSING_DIR to feed a pipeline (liftover, manual + variant entry, classification import) rather than a file a user handed us - so it can be + thrown away with the pipeline that consumed it. WEB_UPLOAD files live under UPLOAD_DIR and + are named via file_field, so they can never match. """ + if not self.path or self.import_source == ImportSource.WEB_UPLOAD: + return False + import_processing_root = os.path.realpath(settings.IMPORT_PROCESSING_DIR) + resolved = os.path.realpath(self.path) + return resolved.startswith(import_processing_root + os.sep) + def can_view(self, user_or_group: Union[User, Group]) -> bool: if isinstance(user_or_group, User): return user_or_group.is_superuser or self.user == user_or_group @@ -208,9 +224,21 @@ def get_pipeline_processing_subdir(self, subdir): return sd def remove_processing_files(self): - pipeline_processing_dir = self.get_pipeline_processing_dir() - logging.info("*** Deleting files for pipeline %d - '%s'", self.pk, pipeline_processing_dir) - shutil.rmtree(pipeline_processing_dir) + remove_import_processing_dir(self.pk) + + def remove_generated_input_file(self): + """ Some pipelines are fed a VCF we wrote for them (liftover, manual variant entry, + classification import) into a scratch dir of its own - once the pipeline is done with it, + nothing else can use it. Files written *inside* pipeline_ (variant tags, TSO500) belong + to remove_processing_files. """ + file_upload = self.file_upload + if not file_upload.is_import_processing_scratch: + return + input_dir = os.path.dirname(os.path.realpath(file_upload.path)) + if os.path.realpath(import_processing_dir_path(self.pk)) == input_dir: + return + logging.info("*** Deleting generated input dir for pipeline %d - '%s'", self.pk, input_dir) + shutil.rmtree(input_dir, ignore_errors=True) def start(self): logging.debug("upload_pipeline.start()") @@ -236,6 +264,7 @@ def success(self, items_processed=None, processing_seconds_wall_time=None, proce create_event(self.file_upload.user, f"import_{self.get_file_type_display()}_success") if settings.IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS: self.remove_processing_files() + self.remove_generated_input_file() def error(self, error_message): # FIXME remove this, cause of error might not be the most recent exception @@ -525,6 +554,23 @@ def __str__(self): return description +@receiver(post_delete, sender=UploadPipeline) +def upload_pipeline_post_delete_handler(sender, instance, **kwargs): # pylint: disable=unused-argument + """ The row is gone, so nothing can name these directories any more - there is no sweep of + IMPORT_PROCESSING_DIR to catch them later (that is what the import_processing_cleanup command + is for, and only for what leaked before #928). + + Unconditional: the setting is about keeping a *successful* pipeline's scratch around to look at, + which needs the pipeline to still be there to look at it from. Using the signal rather than + overriding delete() also covers queryset deletes and FileUpload/VCF cascades - registering a + receiver disables Django's fast-delete path. """ + try: + instance.remove_processing_files() + instance.remove_generated_input_file() + except Exception: # Deleting the row is what matters - a file we can't remove must not block it + logging.exception("Failed removing files for deleted UploadPipeline %s", instance.pk) + + @receiver(pre_delete, sender=UploadedVCF) def pre_delete_uploaded_vcf(sender, instance, *args, **kwargs): if vcf := instance.vcf: diff --git a/upload/tasks/vcf/import_vcf_step_task.py b/upload/tasks/vcf/import_vcf_step_task.py index 1b6b3c77a..3b71f5484 100644 --- a/upload/tasks/vcf/import_vcf_step_task.py +++ b/upload/tasks/vcf/import_vcf_step_task.py @@ -231,6 +231,14 @@ def pipeline_start_task(upload_pipeline_id): @celery.shared_task def pipeline_success_task(upload_pipeline_id): + """ The only thing that takes a VCF pipeline out of PROCESSING - it records the timings, fires the + success event and reclaims the pipeline's scratch dir (@see UploadPipeline.success). + + The PROCESSING guard makes a repeat run a no-op: FINISH can legitimately be scheduled more than + once (two DATA_INSERTION steps finishing together each call check_pipeline_stage), and while + schedule_pipeline_stage_steps de-duplicates the steps on start_date it appends this task + unconditionally. #928: a FINISH step that set SUCCESS itself made the guard swallow the real + close, so none of the above ran. """ upload_pipeline = UploadPipeline.objects.get(pk=upload_pipeline_id) if upload_pipeline.status == ProcessingStatus.PROCESSING: steps = upload_pipeline.uploadstep_set.all() diff --git a/upload/tasks/vcf/import_vcf_tasks.py b/upload/tasks/vcf/import_vcf_tasks.py index c4faeb305..1943e2346 100644 --- a/upload/tasks/vcf/import_vcf_tasks.py +++ b/upload/tasks/vcf/import_vcf_tasks.py @@ -149,16 +149,6 @@ def process_items(self, upload_step: UploadStep): return 0 -class UploadPipelineFinishedTask(ImportVCFStepTask): - - def process_items(self, upload_step): - upload_pipeline = upload_step.upload_pipeline - if upload_pipeline.status == ProcessingStatus.PROCESSING: - upload_pipeline.status = ProcessingStatus.SUCCESS - upload_pipeline.save() - return 0 - - class ImportCreateUploadedVCFTask(ImportVCFStepTask): def process_items(self, upload_step): @@ -279,7 +269,6 @@ def process_vcf_file_task(vcf_filename, name, user_id, import_source): GeneLevelInsertGeneFusionsTask = app.register_task(GeneLevelInsertGeneFusionsTask()) CheckStartAnnotationTask = app.register_task(CheckStartAnnotationTask()) ScheduleMultiFileOutputTasksTask = app.register_task(ScheduleMultiFileOutputTasksTask()) -UploadPipelineFinishedTask = app.register_task(UploadPipelineFinishedTask()) ImportCreateUploadedVCFTask = app.register_task(ImportCreateUploadedVCFTask()) ProcessVCFSetMaxVariantTask = app.register_task(ProcessVCFSetMaxVariantTask()) ProcessVCFLinkAllelesSetMaxVariantTask = app.register_task(ProcessVCFLinkAllelesSetMaxVariantTask()) diff --git a/upload/tests/test_import_processing_cleanup.py b/upload/tests/test_import_processing_cleanup.py new file mode 100644 index 000000000..bf68ab790 --- /dev/null +++ b/upload/tests/test_import_processing_cleanup.py @@ -0,0 +1,83 @@ +""" +The one-off sweep of directories that leaked before #928 - manage.py import_processing_cleanup. + +The rule it has to get right is "only remove what its owner is finished with", so each case here is a +directory whose owner is present-and-done, present-and-still-working, or gone entirely. +""" +import os +import tempfile +import time +from io import StringIO + +from django.contrib.auth.models import User +from django.core.management import call_command +from django.test import TestCase, override_settings + +from annotation.models.models import ManualVariantEntryCollection +from snpdb.models.models_enums import ImportSource +from snpdb.models.models_genome import GenomeBuild +from upload.models import ( + FileUpload, + ProcessingStatus, + UploadedFileTypes, + UploadedManualVariantEntryCollection, + UploadPipeline, +) + +OLD_MTIME = time.time() - 30 * 24 * 60 * 60 + + +class ImportProcessingCleanupTest(TestCase): + + @classmethod + def setUpTestData(cls): + super().setUpTestData() + cls.user = User.objects.create_user(username="import_processing_cleanup_user", password="x") + cls.grch37 = GenomeBuild.get_name_or_alias("GRCh37") + + def _make_pipeline(self, status: str) -> UploadPipeline: + mvec = ManualVariantEntryCollection.objects.create(user=self.user, genome_build=self.grch37) + file_upload = FileUpload.objects.create(user=self.user, name="manual_variant_entry", + path="/nowhere/manual_variant_entry.vcf", + file_type=UploadedFileTypes.MANUAL_VARIANT_ENTRY, + import_source=ImportSource.WEB) + UploadedManualVariantEntryCollection.objects.create(file_upload=file_upload, collection=mvec) + return UploadPipeline.objects.create(status=status, file_upload=file_upload) + + @staticmethod + def _make_dir(import_processing_dir: str, name: str, empty=False) -> str: + path = os.path.join(import_processing_dir, name) + os.makedirs(path) + if not empty: + with open(os.path.join(path, "scratch.txt"), "w") as f: + f.write("x") + os.utime(path, (OLD_MTIME, OLD_MTIME)) + return path + + def test_removes_only_what_owners_are_done_with(self): + successful = self._make_pipeline(ProcessingStatus.SUCCESS) + errored = self._make_pipeline(ProcessingStatus.ERROR) + with tempfile.TemporaryDirectory() as import_processing_dir: + paths = { + "successful": self._make_dir(import_processing_dir, f"pipeline_{successful.pk}"), + "errored": self._make_dir(import_processing_dir, f"pipeline_{errored.pk}"), + "no_row": self._make_dir(import_processing_dir, "pipeline_99999999"), + "clingen_empty": self._make_dir(import_processing_dir, "clingen_allele_registry_abc", empty=True), + "clingen_dump": self._make_dir(import_processing_dir, "clingen_allele_registry_failures"), + "unit_test": self._make_dir(import_processing_dir, "test"), + "unknown": self._make_dir(import_processing_dir, "something_we_dont_know"), + } + recent = os.path.join(import_processing_dir, "pipeline_99999998") + os.makedirs(recent) + + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir): + call_command("import_processing_cleanup", dry_run=True, stdout=StringIO()) + self.assertTrue(all(os.path.exists(p) for p in paths.values()), "--dry-run removed something") + + call_command("import_processing_cleanup", stdout=StringIO()) + + for key in ("successful", "no_row", "clingen_empty", "unit_test"): + self.assertFalse(os.path.exists(paths[key]), f"{key} should have been removed") + for key in ("errored", "clingen_dump", "unknown"): + self.assertTrue(os.path.exists(paths[key]), f"{key} should have been kept") + self.assertTrue(os.path.exists(recent), "an entry modified today should have been kept") diff --git a/upload/tests/test_pipeline_cleanup.py b/upload/tests/test_pipeline_cleanup.py new file mode 100644 index 000000000..61c697646 --- /dev/null +++ b/upload/tests/test_pipeline_cleanup.py @@ -0,0 +1,190 @@ +""" +#928: a VCF pipeline is closed in exactly one place, and its scratch files go with it. + +Covers the two things that leaked - pipeline_success_task being defeated by a FINISH step that set +SUCCESS itself, and nothing removing a pipeline's directories when its row was deleted. +""" +import os +import tempfile +from unittest.mock import patch + +from django.contrib.auth.models import User +from django.test import TestCase, override_settings + +from annotation.models.models import ManualVariantEntryCollection +from library.django_utils.django_file_utils import ( + get_import_processing_dir, + import_processing_dir_path, +) +from snpdb.clingen_allele_api import ClinGenAlleleRegistryAPI +from snpdb.models.models_enums import ImportSource +from snpdb.models.models_genome import GenomeBuild +from upload.models import ( + FileUpload, + ProcessingStatus, + UploadedFileTypes, + UploadedManualVariantEntryCollection, + UploadPipeline, + UploadStep, + UploadStepTaskType, + VCFPipelineStage, +) +from upload.tasks.vcf.import_vcf_step_task import pipeline_success_task +from upload.uploaded_file_type import retry_upload_pipeline + + +class ImportProcessingDirTest(TestCase): + + def test_path_helper_creates_nothing(self): + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir): + path = import_processing_dir_path(42) + self.assertFalse(os.path.exists(path)) + self.assertTrue(os.path.exists(get_import_processing_dir(42))) + + def test_clingen_api_creates_nothing(self): + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir): + ClinGenAlleleRegistryAPI() + self.assertEqual(os.listdir(import_processing_dir), []) + + +class PipelineCleanupTest(TestCase): + """ Manual variant entry is the smallest pipeline whose input VCF we generate ourselves """ + + @classmethod + def setUpTestData(cls): + super().setUpTestData() + cls.user = User.objects.create_user(username="pipeline_cleanup_user", password="x") + cls.grch37 = GenomeBuild.get_name_or_alias("GRCh37") + + def _make_pipeline(self, generated_input=True) -> UploadPipeline: + """ Writes the generated input VCF into its own manual_variants_ dir, as + annotation.manual_variant_entry.create_manual_variants does """ + mvec = ManualVariantEntryCollection.objects.create(user=self.user, genome_build=self.grch37) + if generated_input: + working_dir = get_import_processing_dir(mvec.pk, "manual_variants") + path = os.path.join(working_dir, "manual_variant_entry.vcf") + else: + path = os.path.join(tempfile.gettempdir(), "somebody_elses.vcf") + with open(path, "w") as f: + f.write("##fileformat=VCFv4.2\n") + + file_upload = FileUpload.objects.create(user=self.user, name="manual_variant_entry", path=path, + file_type=UploadedFileTypes.MANUAL_VARIANT_ENTRY, + import_source=ImportSource.WEB) + UploadedManualVariantEntryCollection.objects.create(file_upload=file_upload, collection=mvec) + upload_pipeline = UploadPipeline.objects.create(status=ProcessingStatus.PROCESSING, + file_upload=file_upload) + # Something in the pipeline's own dir, so we can see it go + with open(os.path.join(upload_pipeline.get_pipeline_processing_dir(), "split_1.vcf"), "w") as f: + f.write("") + return upload_pipeline + + def _make_finish_step(self, upload_pipeline: UploadPipeline) -> UploadStep: + return UploadStep.objects.create(upload_pipeline=upload_pipeline, + name="LiftoverCompleteTask", + sort_order=1, + task_type=UploadStepTaskType.CELERY, + pipeline_stage_dependency=VCFPipelineStage.FINISH, + script="upload.tasks.vcf.import_vcf_tasks.LiftoverCompleteTask", + start_date="2026-09-08T00:00:00Z", + end_date="2026-09-08T00:01:00Z") + + def test_success_closes_pipeline_and_removes_files(self): + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir): + upload_pipeline = self._make_pipeline() + self._make_finish_step(upload_pipeline) + pipeline_dir = import_processing_dir_path(upload_pipeline.pk) + + pipeline_success_task(upload_pipeline.pk) + + upload_pipeline.refresh_from_db() + self.assertEqual(upload_pipeline.status, ProcessingStatus.SUCCESS) + self.assertEqual(upload_pipeline.processing_seconds_wall_time, 60) + self.assertFalse(os.path.exists(pipeline_dir)) + self.assertFalse(os.path.exists(upload_pipeline.file_upload.path)) + + def test_success_keeps_files_when_setting_off(self): + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir, + IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS=False): + upload_pipeline = self._make_pipeline() + self._make_finish_step(upload_pipeline) + + pipeline_success_task(upload_pipeline.pk) + + upload_pipeline.refresh_from_db() + self.assertEqual(upload_pipeline.status, ProcessingStatus.SUCCESS) + self.assertTrue(os.path.exists(import_processing_dir_path(upload_pipeline.pk))) + self.assertTrue(os.path.exists(upload_pipeline.file_upload.path)) + + def test_success_task_is_a_no_op_once_closed(self): + """ FINISH can be scheduled more than once, so the second run has to leave a closed pipeline + alone rather than re-fire its success event """ + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir): + upload_pipeline = self._make_pipeline() + self._make_finish_step(upload_pipeline) + pipeline_success_task(upload_pipeline.pk) + UploadPipeline.objects.filter(pk=upload_pipeline.pk).update(items_processed=123) + + pipeline_success_task(upload_pipeline.pk) + + upload_pipeline.refresh_from_db() + self.assertEqual(upload_pipeline.status, ProcessingStatus.SUCCESS) + self.assertEqual(upload_pipeline.items_processed, 123) + + def test_delete_removes_files(self): + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir, + IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS=False): + upload_pipeline = self._make_pipeline() + pipeline_dir = import_processing_dir_path(upload_pipeline.pk) + input_path = upload_pipeline.file_upload.path + + upload_pipeline.delete() + + self.assertFalse(os.path.exists(pipeline_dir)) + self.assertFalse(os.path.exists(input_path)) + + def test_file_upload_cascade_removes_files(self): + """ Deleting the FileUpload (eg a user deleting their upload) cascades to the pipeline """ + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir): + upload_pipeline = self._make_pipeline() + pipeline_dir = import_processing_dir_path(upload_pipeline.pk) + input_path = upload_pipeline.file_upload.path + + upload_pipeline.file_upload.delete() + + self.assertFalse(os.path.exists(pipeline_dir)) + self.assertFalse(os.path.exists(input_path)) + + def test_delete_keeps_a_file_we_did_not_generate(self): + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir): + upload_pipeline = self._make_pipeline(generated_input=False) + input_path = upload_pipeline.file_upload.path + try: + upload_pipeline.delete() + self.assertTrue(os.path.exists(input_path)) + finally: + os.remove(input_path) + + def test_retry_keeps_the_generated_input(self): + """ Retry re-runs the pipeline against the VCF we generated for it, so only the pipeline's own + working files go """ + with tempfile.TemporaryDirectory() as import_processing_dir: + with override_settings(IMPORT_PROCESSING_DIR=import_processing_dir): + upload_pipeline = self._make_pipeline() + pipeline_dir = import_processing_dir_path(upload_pipeline.pk) + input_path = upload_pipeline.file_upload.path + + with patch("upload.uploaded_file_type.process_upload_pipeline") as mock_process: + mock_process.return_value = (upload_pipeline,) + retry_upload_pipeline(upload_pipeline) + + self.assertFalse(os.path.exists(pipeline_dir)) + self.assertTrue(os.path.exists(input_path)) diff --git a/variantgrid/settings/components/default_settings.py b/variantgrid/settings/components/default_settings.py index 272780a7f..44da656ad 100644 --- a/variantgrid/settings/components/default_settings.py +++ b/variantgrid/settings/components/default_settings.py @@ -10,6 +10,7 @@ import re import socket import sys +import tempfile from collections import defaultdict from library.django_utils.django_secret_key import get_or_create_django_secret_key @@ -447,6 +448,11 @@ PATIENTS_API_EXTERNAL_MANAGER_CREATE_ADMIN_ONLY = True IMPORT_PROCESSING_DIR = os.path.join(PRIVATE_DATA_ROOT, 'import_processing') IMPORT_PROCESSING_DELETE_TEMP_FILES_ON_SUCCESS = True +if UNIT_TEST: + # A test that builds an UploadPipeline writes pipeline_ wherever this points, and test-db + # pks collide with the dev database's - so keep the suite out of the real tree entirely (#928). + # VariantGridTestRunner.teardown_test_environment removes this + IMPORT_PROCESSING_DIR = os.path.join(tempfile.gettempdir(), "variantgrid_unit_test_import_processing") # Where partition dump files are written when an archivable model (VAV, ClinVarVersion, CohortGenotypeCollection, ...) # is archived via the pre-drop archival pipeline (#1537). diff --git a/variantgrid/templates/default_templates/changelog.html b/variantgrid/templates/default_templates/changelog.html index d74d6a457..507a4bcbc 100644 --- a/variantgrid/templates/default_templates/changelog.html +++ b/variantgrid/templates/default_templates/changelog.html @@ -285,6 +285,11 @@

Patients / Samples

  • private#933 - Cohort page - tick samples to launch a Duo/Trio/Quad or spawn a sub cohort
+

VCF Upload

+
    +
  • #928 - Imports record their timings and clean up their temp files again (a finish step was closing the pipeline early)
    • +
  • #928 - Deleting an import removes its temp files; new import_processing_cleanup command reclaims old ones
    • +
diff --git a/variantgrid/test_runner.py b/variantgrid/test_runner.py index 3f7ad66db..5e6ef5bbd 100644 --- a/variantgrid/test_runner.py +++ b/variantgrid/test_runner.py @@ -1,8 +1,10 @@ import atexit import json import os +import shutil import cdot.hgvs.dataproviders.fasta_seqfetcher as fasta_seqfetcher +from django.conf import settings from django.db import connections from django.db.migrations.loader import MigrationLoader from django.test.runner import DiscoverRunner @@ -25,6 +27,12 @@ def setup_test_environment(self, **kwargs): ClinGenAlleleRegistryAPI.override_class = MockClinGenAlleleRegistryAPI TranscriptSequenceFetcher.override_class = MockTranscriptSequenceFetcher + def teardown_test_environment(self, **kwargs): + super().teardown_test_environment(**kwargs) + # settings.IMPORT_PROCESSING_DIR is a per-suite temp dir (see default_settings UNIT_TEST block), + # so whatever the run left in it - a pipeline dir, fake annotation scratch - goes with it + shutil.rmtree(settings.IMPORT_PROCESSING_DIR, ignore_errors=True) + def setup_databases(self, **kwargs): if self.keepdb and self.parallel > 1: self._drop_test_db_clones()