Create a cloud function that processes an event from cloud pubsub, and transforms the JSON data using a parser, and upserts that data into metamist. (#532)

* Refactoring ETL folder, preparing first draft of etl_post and etl_load functions.

* Sample Json Parser sceleton class.

* Prepared test file for SampleJsonParser, defined simple parse function.

* Fixing linting issues.

* Added pytest to requirements-dev.txt.

* Fixing small linting issues.

* Update metamist_infrastructure setup to contain new cloud function.

* Rename post to extract, implement Pub/Sub functionality in the ETL driver.

* ETL - load func implemented pub/sub type of payload, cleanup driver class.

* Implement simple unit tests for etl/load and etl/extract.

* Update setup and add docstring to test functions.

* Move ETL unit tests to own directory and update workflow script to include them.

* Update readme for metamist_infra, update requirements-dev and ignore to be able to test pulumi up command on local project.

* Remove cpg_infra dependency from requirements-dev.

* Implement slack notification, update readme.

* Integrated ETL load function with GenericMetadataParser, using temp metamist.tar.gz approach before new version of metamist is released.

* Updated unit tests for etl functions, added mock for GenericMetadataParser.

* Added metamist.tar.gz file to cloud function assets.

* Integrated with DEV metamist, using dry_run=True for time being.

* Fixed linting issues.

* ETL load function - updating async run

* Kept parser original formatting.

* Added comments and small changes to reflect the pull request feedback.

* Added etl log table, implemented matching parser to url path.

* Updade ETL load to load available parser dynamically, fully implemented detailed Slack messaging.

* Fixed linting issues.

* Implemented entry points for metamist_parser, added ability to include extra private python repos in gc functions.

* Linting

* Added default ETL load config.

* Resolving conflicts with dev branch.

* Fixing linting issues after merging from dev.

* Actioned pull request feedback.

* Added more comments regarding serverless-robot-prod account.

* bump mypy.ini to 3.11, small cleanup.

* revert back mypy to 3.10.

* bumping metamist_infra version, fixing import isssues, using new cpg_infra archive_folder implementation.

* Integrate with latest cpg-infra changes, esp. constructing private repo url from ArtifactRegistry.

* Cleanup unused code.

* Makong Enums as string.

* Changes as per review request.

* Bump version: 6.2.1 → 6.3.0

* Removing metamist zip.

* Regenerating package lock.

---------

Co-authored-by: Milo Hyben <[email protected]>

Author: milo-hyben

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 6.2.1
+current_version = 6.3.0
 commit = True
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>[A-z0-9-]+)

.github/workflows/test.yaml

@@ -66,7 +66,7 @@ jobs:
       - name: "Run unit tests"
         id: runtests
         run: |
-          coverage run -m unittest discover -s test/
+          coverage run -m unittest discover -p 'test*.py' -s '.'
           rc=$?
           coverage xml
 

.gitignore

@@ -56,3 +56,6 @@ web/src/__generated__
 
 # mypy
 .mypy_cache/
+
+# pulumi config files
+Pulumi*.yaml

api/server.py

@@ -20,7 +20,7 @@
 from api.settings import PROFILE_REQUESTS, SKIP_DATABASE_CONNECTION
 
 # This tag is automatically updated by bump2version
-_VERSION = '6.2.1'
+_VERSION = '6.3.0'
 
 logger = get_logger()
 

deploy/python/version.txt

@@ -1 +1 @@
-6.2.1
+6.3.0

etl/README.md

@@ -0,0 +1,133 @@
+# ETL
+
+Cloud Functions for ETL.
+
+## ETL_POST
+
+etl_post function accepts json payload. It will insert new record in BigQuery table defined by env variable BIGQUERY_TABLE and pushes new message to PUBSUB_TOPIC
+
+
+## ETL_LOAD
+
+etl_load function expects "request_id" in the payload. It is setup as push subscriber to PUBSUB_TOPIC.
+
+
+## How to test locally
+
+Please use your personal dev project as `$PROJECT_NAME`.
+
+### 1. Setup your environment
+
+```bash
+# setup gcloud authentication
+gcloud auth application-default login
+
+export PROJECT_NAME="gcp-project-name"
+export BIGQUERY_TABLE="$PROJECT_NAME.metamist.etl-data"
+export BIGQUERY_LOG_TABLE="$PROJECT_NAME.metamist.etl-logs"
+export PUBSUB_TOPIC="projects/$PROJECT_NAME/topics/etl-topic"
+
+# setup to run local version of sample-metadata
+export SM_ENVIRONMENT=local
+```
+
+### 2. Create BQ table "$PROJECT_NAME.metamist.etl-data"
+
+### 3. Create TOPIC "projects/$PROJECT_NAME/topics/etl-topic"
+
+### 4. Setup python env
+
+```bash
+cd post
+python3 -m venv env
+source env/bin/activate
+pip install -r requirements.txt
+```
+
+### 5. Start EXTRACT Fun locally
+
+```bash
+functions-framework-python --target etl_extract --debug
+```
+
+### 6. Call etl_extract
+
+```bash
+curl -X 'POST' \
+  'http://localhost:8080/' \
+  -H 'Content-Type: application/json' \
+  -H "Authorization: Bearer $(gcloud auth print-identity-token)" \
+  -d '{"identifier": "AB0002", "name": "j smith", "age": 50, "measurement": "98.7", "observation": "B++", "receipt_date": "1/02/2023"}'
+```
+
+Should return something like this:
+
+```bash
+{
+  "id": "76263e55-a869-4604-afe2-441d9c20221e",
+  "success": true
+}
+```
+
+### 7. Start LOAD Fun locally
+
+Repeat Step 4 inside folder load
+
+```bash
+functions-framework-python --target etl_load --debug
+```
+
+### 8. Call etl_load
+
+Replace request_id with the id returned in Step 6
+
+```bash
+curl -X 'POST' \
+  'http://localhost:8080/' \
+  -H "Authorization: Bearer $(gcloud auth print-identity-token)" \
+  -d '{"request_id": "76263e55-a869-4604-afe2-441d9c20221e"}'
+```
+
+Should return something like this:
+
+```bash
+{
+  "id": "76263e55-a869-4604-afe2-441d9c20221e",
+  "success": true
+}
+```
+
+
+### 9. Deploy functions for testing on the cloud
+
+```bash
+cd ../load
+
+gcloud functions deploy etl_load \
+    --gen2 \
+    --runtime=python311 \
+    --project=$PROJECT_NAME \
+    --region=australia-southeast1 \
+    --source=. \
+    --entry-point=etl_load \
+    --trigger-http \
+    --no-allow-unauthenticated \
+    --set-env-vars BIGQUERY_TABLE=$BIGQUERY_TABLE \
+    --set-env-vars PUBSUB_TOPIC=$PUBSUB_TOPIC
+```
+
+```bash
+cd ../extract
+
+gcloud functions deploy etl_extract \
+    --gen2 \
+    --runtime=python311 \
+    --project=$PROJECT_NAME \
+    --region=australia-southeast1 \
+    --source=. \
+    --entry-point=etl_post \
+    --trigger-http \
+    --no-allow-unauthenticated \
+    --set-env-vars BIGQUERY_TABLE=$BIGQUERY_TABLE \
+    --set-env-vars PUBSUB_TOPIC=$PUBSUB_TOPIC
+```

etl/bq_log_schema.json

@@ -0,0 +1,26 @@
+[
+    {
+        "name": "request_id",
+        "type": "STRING",
+        "mode": "REQUIRED",
+        "description": "Unique UUID for the row"
+    },
+    {
+        "name": "timestamp",
+        "type": "TIMESTAMP",
+        "mode": "REQUIRED",
+        "description": "Timestamp of processing the row"
+    },
+    {
+        "name": "status",
+        "type": "STRING",
+        "mode": "REQUIRED",
+        "description": "Status of the processing FAILED/SUCCESS"
+    },
+    {
+        "name": "details",
+        "type": "JSON",
+        "mode": "REQUIRED",
+        "description": "Output the processing"
+    }
+]

etl/endpoint/__init__.py renamed to etl/extract/__init__.py

@@ -8,17 +8,15 @@
 import functions_framework
 import google.cloud.bigquery as bq
 from cpg_utils.cloud import email_from_id_token
+
 from google.cloud import pubsub_v1  # type: ignore
 
 BIGQUERY_TABLE = os.getenv('BIGQUERY_TABLE')
 PUBSUB_TOPIC = os.getenv('PUBSUB_TOPIC')
 
-_BQ_CLIENT = bq.Client()
-_PUBSUB_CLIENT = pubsub_v1.PublisherClient()
-
 
 @functions_framework.http
-def etl_post(request: flask.Request):
+def etl_extract(request: flask.Request):
     """HTTP Cloud Function.
     Args:
         request (flask.Request): The request object.
@@ -31,6 +29,20 @@ def etl_post(request: flask.Request):
         For more information on how Flask integrates with Cloud
         Functions, see the `Writing HTTP functions` page.
         <https://cloud.google.com/functions/docs/writing/http#http_frameworks>
+
+
+    Example of payload:
+
+    json_data = {
+        'sample_id': '123456',
+        'external_id': 'GRK100311',
+        'individual_id': '608',
+        'sequencing_type': 'exome',
+        'collection_centre': 'KCCG',
+        'collection_date': '2023-08-05T01:39:28.611476',
+        'collection_specimen': 'blood'
+    }
+
     """
 
     auth = request.authorization
@@ -40,6 +52,7 @@ def etl_post(request: flask.Request):
     request_id = str(uuid.uuid4())
 
     jbody = request.json
+
     if callable(jbody):
         # request.json is it in reality, but the type checker is saying it's callable
         jbody = jbody()
@@ -59,11 +72,14 @@ def etl_post(request: flask.Request):
         'timestamp': datetime.datetime.utcnow().isoformat(),
         'type': request.path,
         'submitting_user': email_from_id_token(auth.token),
-        'body': jbody_str,
     }
 
     # throw an exception if one occurs
-    errors = _BQ_CLIENT.insert_rows_json(BIGQUERY_TABLE, [bq_obj])
+    bq_client = bq.Client()
+    pubsub_client = pubsub_v1.PublisherClient()
+
+    errors = bq_client.insert_rows_json(BIGQUERY_TABLE, [bq_obj | {'body': jbody_str}])
+
     if errors:
         return {
             'success': False,
@@ -72,8 +88,10 @@ def etl_post(request: flask.Request):
         }, 500
 
     # publish to pubsub
+    # message contains all the attributes except body which can be large
+    # and already stored in BQ table
     try:
-        _PUBSUB_CLIENT.publish(PUBSUB_TOPIC, json.dumps(bq_obj).encode())
+        pubsub_client.publish(PUBSUB_TOPIC, json.dumps(bq_obj).encode())
     except Exception as e:  # pylint: disable=broad-exception-caught
         logging.error(f'Failed to publish to pubsub: {e}')