Merge pull request #695 from broadinstitute/benb/fix_logging

Update Logging.

Author: bpblanken

v03_pipeline/lib/logger.py

@@ -0,0 +1,35 @@
+import logging.config
+
+LOG_CONFIG = {
+    'version': 1,
+    'disable_existing_loggers': True,
+    'propagate': True,
+    'formatters': {
+        'default': {
+            'format': '%(asctime)s - %(module)s - %(levelname)s - %(message)s',
+            'datefmt': '%Y-%m-%d %H:%M:%S',
+        },
+    },
+    'handlers': {
+        'default': {
+            'formatter': 'default',
+            'class': 'logging.StreamHandler',
+        },
+    },
+    'loggers': {
+        '': {
+            'level': 'INFO',
+            'handlers': ['default'],
+            'propagate': True,
+        },
+    },
+}
+
+_CONFIGURED = False
+
+def get_logger(name: str):
+    global _CONFIGURED # noqa: PLW0603
+    if not _CONFIGURED:
+        logging.config.dictConfig(LOG_CONFIG)
+        _CONFIGURED = True
+    return logging.getLogger(name)

v03_pipeline/lib/reference_data/clinvar.py

@@ -6,6 +6,7 @@
 import hail as hl
 
 from v03_pipeline.lib.annotations.enums import CLINVAR_PATHOGENICITIES_LOOKUP
+from v03_pipeline.lib.logger import get_logger
 from v03_pipeline.lib.model.definitions import ReferenceGenome
 
 CLINVAR_ASSERTIONS = [
@@ -33,6 +34,7 @@
     },
 )
 
+logger = get_logger(__name__)
 
 def safely_move_to_gcs(tmp_file_name, gcs_tmp_file_name):
     try:
@@ -45,8 +47,8 @@ def safely_move_to_gcs(tmp_file_name, gcs_tmp_file_name):
             ],
             check=True,
         )
-    except subprocess.CalledProcessError as e:
-        print(e)
+    except subprocess.CalledProcessError:
+        logger.exception(f'Failed to move local tmp file {tmp_file_name} to gcs')
 
 
 def parsed_clnsig(ht: hl.Table):

v03_pipeline/lib/reference_data/compare_globals.py

@@ -1,8 +1,8 @@
 import dataclasses
-import logging
 
 import hail as hl
 
+from v03_pipeline.lib.logger import get_logger
 from v03_pipeline.lib.model import (
     DatasetType,
     ReferenceDatasetCollection,
@@ -16,7 +16,7 @@
     parse_dataset_version,
 )
 
-logger = logging.getLogger(__name__)
+logger = get_logger(__name__)
 
 
 @dataclasses.dataclass

v03_pipeline/lib/tasks/base/base_hail_table_task.py

@@ -0,0 +1,48 @@
+import hail as hl
+import luigi
+
+from v03_pipeline.lib.logger import get_logger
+from v03_pipeline.lib.model import DatasetType, Env, ReferenceGenome, SampleType
+from v03_pipeline.lib.tasks.files import GCSorLocalFolderTarget
+
+logger = get_logger(__name__)
+
+class BaseHailTableTask(luigi.Task):
+    reference_genome = luigi.EnumParameter(enum=ReferenceGenome)
+    dataset_type = luigi.EnumParameter(enum=DatasetType)
+    sample_type = luigi.EnumParameter(enum=SampleType)
+
+    def output(self) -> luigi.Target:
+        raise NotImplementedError
+
+    def complete(self) -> bool:
+        return GCSorLocalFolderTarget(self.output().path).exists()
+
+    def init_hail(self):
+        # Need to use the GCP bucket as temp storage for very large callset joins
+        hl.init(tmp_dir=Env.HAIL_TMPDIR, idempotent=True)
+
+        # Interval ref data join causes shuffle death, this prevents it
+        hl._set_flags(use_new_shuffle='1', no_whole_stage_codegen='1')  # noqa: SLF001
+
+# NB: these are defined over luigi.Task instead of the BaseHailTableTask so that
+# they work on file dependencies.
+@luigi.Task.event_handler(luigi.Event.DEPENDENCY_MISSING)
+def dependency_missing(task):
+    logger.info(f'{task} dependency_missing')
+
+@luigi.Task.event_handler(luigi.Event.DEPENDENCY_PRESENT)
+def dependency_present(task):
+    logger.info(f'{task} dependency_present')
+
+@luigi.Task.event_handler(luigi.Event.START)
+def start(task):
+    logger.info(f'{task} start')
+
+@luigi.Task.event_handler(luigi.Event.FAILURE)
+def failure(task, _):
+    logger.exception(f'{task} failure')
+
+@luigi.Task.event_handler(luigi.Event.SUCCESS)
+def success(task):
+    logger.info(f'{task} success')

v03_pipeline/lib/tasks/base/base_update_task.py

@@ -1,29 +1,10 @@
 import hail as hl
-import luigi
 
 from v03_pipeline.lib.misc.io import write
-from v03_pipeline.lib.model import DatasetType, Env, ReferenceGenome, SampleType
-from v03_pipeline.lib.tasks.files import GCSorLocalFolderTarget
+from v03_pipeline.lib.tasks.base.base_hail_table_task import BaseHailTableTask
 
 
-class BaseUpdateTask(luigi.Task):
-    reference_genome = luigi.EnumParameter(enum=ReferenceGenome)
-    dataset_type = luigi.EnumParameter(enum=DatasetType)
-    sample_type = luigi.EnumParameter(enum=SampleType)
-
-    def output(self) -> luigi.Target:
-        raise NotImplementedError
-
-    def complete(self) -> bool:
-        return GCSorLocalFolderTarget(self.output().path).exists()
-
-    def init_hail(self):
-        # Need to use the GCP bucket as temp storage for very large callset joins
-        hl.init(tmp_dir=Env.HAIL_TMPDIR, idempotent=True)
-
-        # Interval ref data join causes shuffle death, this prevents it
-        hl._set_flags(use_new_shuffle='1', no_whole_stage_codegen='1')  # noqa: SLF001
-
+class BaseUpdateTask(BaseHailTableTask):
     def run(self) -> None:
         self.init_hail()
         if not self.output().exists():

v03_pipeline/lib/tasks/base/base_write_task.py

@@ -1,29 +1,10 @@
 import hail as hl
-import luigi
 
 from v03_pipeline.lib.misc.io import write
-from v03_pipeline.lib.model import DatasetType, Env, ReferenceGenome, SampleType
-from v03_pipeline.lib.tasks.files import GCSorLocalFolderTarget
+from v03_pipeline.lib.tasks.base.base_hail_table_task import BaseHailTableTask
 
 
-class BaseWriteTask(luigi.Task):
-    reference_genome = luigi.EnumParameter(enum=ReferenceGenome)
-    dataset_type = luigi.EnumParameter(enum=DatasetType)
-    sample_type = luigi.EnumParameter(enum=SampleType)
-
-    def output(self) -> luigi.Target:
-        raise NotImplementedError
-
-    def complete(self) -> bool:
-        return GCSorLocalFolderTarget(self.output().path).exists()
-
-    def init_hail(self):
-        # Need to use the GCP bucket as temp storage for very large callset joins
-        hl.init(tmp_dir=Env.HAIL_TMPDIR, idempotent=True)
-
-        # Interval ref data join causes shuffle death, this prevents it
-        hl._set_flags(use_new_shuffle='1', no_whole_stage_codegen='1')  # noqa: SLF001
-
+class BaseWriteTask(BaseHailTableTask):
     def run(self) -> None:
         self.init_hail()
         ht = self.create_table()

v03_pipeline/lib/tasks/reference_data/updated_reference_dataset_collection.py

@@ -1,8 +1,7 @@
-import logging
-
 import hail as hl
 import luigi
 
+from v03_pipeline.lib.logger import get_logger
 from v03_pipeline.lib.model import ReferenceDatasetCollection
 from v03_pipeline.lib.paths import valid_reference_dataset_collection_path
 from v03_pipeline.lib.reference_data.compare_globals import (
@@ -15,7 +14,7 @@
 from v03_pipeline.lib.tasks.base.base_update_task import BaseUpdateTask
 from v03_pipeline.lib.tasks.files import GCSorLocalTarget
 
-logger = logging.getLogger(__name__)
+logger = get_logger(__name__)
 
 
 class UpdatedReferenceDatasetCollectionTask(BaseUpdateTask):

v03_pipeline/lib/tasks/write_project_family_tables_test.py

@@ -32,7 +32,6 @@ def test_snv_write_project_family_tables_task(self) -> None:
             hl.read_table(write_family_table_task.output().path)
             for write_family_table_task in write_project_family_tables.dynamic_write_family_table_tasks
         ]
-        print('bensadf', len(hts))
         self.assertCountEqual(
             [ht.globals.sample_ids.collect() for ht in hts],
             [