Merge pull request #290 from clingen-data-model/project-labels

Project labels and global exception handler

Author: toneillbroad

clinvar_ingest/cloud/bigquery/stored_procedures.py

@@ -31,7 +31,7 @@
     "CALL `clinvar_ingest.dataset_preparation`({0});",
     "CALL `clinvar_ingest.temporal_data_collection`({0});",
     "CALL `clinvar_ingest.temporal_data_summation`();",
-    "CALL `clinvar_ingest.variation_tracker`();",
+    "CALL `clinvar_ingest.tracker_report_update`();",
 ]
 
 

misc/bin/bq_ingester.py

@@ -57,6 +57,8 @@ def _get_bq_client() -> bigquery.Client:
         setattr(_get_bq_client, "client", bigquery.Client())
     return getattr(_get_bq_client, "client")
 
+################################################################
+### rollback exception handler for deleting processing_history entries
 
 rollback_rows = []
 

misc/bin/deploy-job.sh

@@ -127,11 +127,24 @@ env_vars="$env_vars,BQ_DEST_PROJECT=${BQ_DEST_PROJECT}"
 # if instance_name contains stored-procedures make env vars
 if [[ $instance_name =~ ^.*stored-procedures.*$ ]]; then
     # Resetting env_vars here - not inheriting previous
-    env_vars="BQ_DEST_PROJECT=${BQ_DEST_PROJECT},CLINVAR_INGEST_BQ_META_DATASET=${CLINVAR_INGEST_BQ_META_DATASET}"
+    env_vars="file_format=${file_format}"
+    env_vars="$env_vars,BQ_DEST_PROJECT=${BQ_DEST_PROJECT}"
+    env_vars="$env_vars,CLINVAR_INGEST_BQ_META_DATASET=${CLINVAR_INGEST_BQ_META_DATASET}"
     env_vars="$env_vars,CLINVAR_INGEST_RELEASE_TAG=${CLINVAR_INGEST_RELEASE_TAG}"
 fi
 
-### TODO - stored-procedures
+####
+# gcloud project level labels must exist before use
+####
+metadata=`gcloud projects describe $project --format=json`
+service=`echo $metadata | jq '.labels["service"]'`
+component=`echo $metadata | jq '.labels["component"]'`
+if [ $service == "null" || $component == "null" ]; then
+  # set default labels on the project
+  gcloud projects update $project --update-labels=service="service",component="component"
+fi
+#####
+
 gcloud run jobs $command $instance_name \
       --cpu=2 \
       --memory=8Gi \
@@ -141,7 +154,8 @@ gcloud run jobs $command $instance_name \
       --command="$clinvar_ingest_cmd" \
       --service-account=$pipeline_service_account \
       --set-env-vars=$env_vars \
-      --set-secrets=CLINVAR_INGEST_SLACK_TOKEN=clinvar-ingest-slack-token:latest
+      --set-secrets=CLINVAR_INGEST_SLACK_TOKEN=clinvar-ingest-slack-token:latest \
+      --labels=service="clinvar-ingest",component=$instance_name
 
 if [[ $instance_name =~ ^.*bq-ingest.*$|^.*stored-procedures.*$ ]]; then
     # turn off file globbing
@@ -151,5 +165,6 @@ if [[ $instance_name =~ ^.*bq-ingest.*$|^.*stored-procedures.*$ ]]; then
       --uri=https://${region}-run.googleapis.com/apis/run.googleapis.com/v1/namespaces/${project}/jobs/${instance_name}:run \
       --http-method POST \
       --oauth-service-account-email=$pipeline_service_account \
-      --schedule='*/15 * * * *'
+      --schedule='*/15 * * * *' \
+      --labels=service="clinvar-ingest",component=${instance_name}-scheduler
 fi

misc/bin/stored-procedures-workflow.py

@@ -5,6 +5,7 @@
 
 import logging
 import sys
+import traceback
 
 from google.cloud import bigquery
 
@@ -28,6 +29,42 @@ def _get_bq_client() -> bigquery.Client:
         setattr(_get_bq_client, "client", bigquery.Client())
     return getattr(_get_bq_client, "client")
 
+################################################################
+### rollback exception handler for deleting processing_history entries
+
+rollback_rows = []
+
+
+def sp_rollback_exception_handler(exc_type, exc_value, exc_traceback):
+    """
+    https://docs.python.org/3/library/sys.html#sys.excepthook
+    """
+    exception_details = "".join(
+        traceback.format_exception(exc_type, exc_value, exc_traceback)
+    )
+
+    # Log the exception
+    _logger.error("Uncaught exception:\n%s", exception_details)
+
+    _logger.warning("Rolling back started SP ingest rows.")
+    for row in rollback_rows:
+        _logger.info(f"Rolling back row: {row}")
+        c = processing_history.delete(
+            processing_history_table=row["processing_history_table"],
+            release_tag=row["release_tag"],
+            file_type=row["file_type"],
+            xml_release_date=row["xml_release_date"],
+            client=_get_bq_client(),
+        )
+        _logger.info(f"Deleted {c} rows from processing_history.")
+
+    # Call the default exception handler
+    sys.__excepthook__(exc_type, exc_value, exc_traceback)
+
+# Add the exception handler as the global exception handler.
+# NOTE: this modifies global state and will affect all subsequent exceptions
+# in this script or any other script which imports this script.
+sys.excepthook = sp_rollback_exception_handler
 
 ################################################################
 ### Initialization code
@@ -84,6 +121,17 @@ def _get_bq_client() -> bigquery.Client:
         """
     _logger.info(msg)
 
+    # Add the started row to the rollback list
+    rollback_rows.append(
+        {
+            "processing_history_table": processing_history_table,
+            "release_tag": vcv_pipeline_version,
+            "file_type": env.file_format_mode,
+            "xml_release_date": str(vcv_xml_release_date),
+            "client": _get_bq_client(),
+        }
+    )
+
 # Now process individual rows
 for row in rows_to_ingest:
     _logger.info(row)
@@ -116,21 +164,19 @@ def _get_bq_client() -> bigquery.Client:
         _logger.info(msg)
         send_slack_message(msg)
     except Exception as e:
-        processing_history.write_started(
-            processing_history_table=processing_history_table,
-            release_date=None,
-            release_tag=vcv_pipeline_version,
-            schema_version=schema_version,
-            file_type=ClinVarIngestFileFormat(env.file_format_mode),
-            client=_get_bq_client(),
-            bucket_dir=vcv_bucket_dir,
-            xml_release_date=str(vcv_xml_release_date),
-            error_if_exists=False,
-        )
         msg = f"""
               Stored procedure execution failed for release dated {vcv_xml_release_date} version
               {vcv_pipeline_version}.
               """
         _logger.error(msg)
         send_slack_message(msg)
         raise e
+
+
+    # Remove the started row from the rollback list, since this ingest has succeeded
+    rollback_rows = [
+        row
+        for row in rollback_rows
+        if row["xml_release_date"] != str(vcv_xml_release_date)
+           or row["release_tag"] != vcv_pipeline_version
+    ]

misc/bin/workflow.py

@@ -1,8 +1,8 @@
 #!/usr/bin/env python3
-import argparse
 import logging
 import os
 import sys
+import traceback
 from pathlib import Path, PurePosixPath
 
 from google.cloud import bigquery
@@ -67,6 +67,44 @@ def _get_bq_client() -> bigquery.Client:
         _get_bq_client.client = bigquery.Client()
     return _get_bq_client.client
 
+################################################################
+### rollback exception handler for deleting processing_history entries
+
+rollback_rows = []
+
+def workflow_rollback_exception_handler(exc_type, exc_value, exc_traceback):
+    """
+    https://docs.python.org/3/library/sys.html#sys.excepthook
+    """
+
+    exception_details = "".join(
+        traceback.format_exception(exc_type, exc_value, exc_traceback)
+    )
+
+    # Log the exception
+    _logger.error("Uncaught exception:\n%s", exception_details)
+
+
+    _logger.warning("Rolling back started processing_history rows.")
+    for row in rollback_rows:
+        _logger.info(f"Rolling back row: {row}")
+        c = processing_history.delete(
+            processing_history_table=row["processing_history_table"],
+            release_tag=row["release_tag"],
+            file_type=row["file_type"],
+            xml_release_date=row["xml_release_date"],
+            client=_get_bq_client(),
+        )
+        _logger.info(f"Deleted {c} rows from processing_history.")
+
+    # Call the default exception handler
+    sys.__excepthook__(exc_type, exc_value, exc_traceback)
+
+
+# Add the exception handler as the global exception handler.
+# NOTE: this modifies global state and will affect all subsequent exceptions
+# in this script or any other script which imports this script.
+sys.excepthook = workflow_rollback_exception_handler
 
 ################################################################
 ### Initialization code
@@ -140,6 +178,17 @@ def _get_bq_client() -> bigquery.Client:
     error_if_exists=False,
 )
 
+# Add the started row to the rollback list
+rollback_rows.append(
+    {
+        "processing_history_table": processing_history_table,
+        "release_tag": env.release_tag,
+        "file_type": file_mode,
+        "xml_release_date": str(release_date),
+        "client": _get_bq_client(),
+    }
+)
+
 
 ################################################################
 # Run copy step. Copies a source XML file from an HTTP/FTP server to GCS
@@ -247,7 +296,7 @@ def parse(payload: ParseRequest, limit=None) -> ParseResponse:
 try:
     parse_response = parse(
         ParseRequest(input_path=copy_response.gcs_path),
-        # limit=1000,
+        #limit=1000,
     )
     _logger.info(f"Parse response: {parse_response.model_dump_json}")
 except Exception as e:
@@ -279,3 +328,11 @@ def parse(payload: ParseRequest, limit=None) -> ParseResponse:
 ################################################################
 _logger.info("Workflow succeeded")
 send_slack_message(workflow_id_message + " - Workflow succeeded.")
+
+# Remove the started row from the rollback list, since this ingest has succeeded
+rollback_rows = [
+    row
+    for row in rollback_rows
+    if row["xml_release_date"] != release_date
+    or row["release_tag"] != env.release_tag
+]