Merge pull request #1 from unity-sds/cmr-trigger

research and prototype CMR trigger options: scheduled CMR queries

Author: galenatjpl

README.md

@@ -46,7 +46,7 @@ Trigger events by themselves don't automatically mean that SDS processing is rea
 As described by [Hua et al. [2022]](#1):
 > A fundamental capability of an SDS is to systematically process science data through a series of data transformations from raw instrument data to geophysical measurements. Data are first made available to the SDS from GDS to be processed to higher level data products. The data transformation steps may utilize ancillary and auxiliary files as well as production rules that stipulate conditions for when each step should be executed.
 
-In an SDS, evaluators are functions (irrespective of how they are deployed and called) that perform adaptation-specific evaluation to determine if the next step in the processing pipeline is ready for execution. 
+In an SDS, evaluators are functions (irrespective of how they are deployed and called) that perform adaptation-specific evaluation to determine if the next step in the processing pipeline is ready for execution.
 
 As an example, the following shows the input-output diagram for the NISAR L-SAR L0B PGE (a.k.a. science algorithm):
 
@@ -113,7 +113,7 @@ and a trigger event payload for a new file that was triggered:
 
 The router will iterate over the set of url configs and attempt to match the URL against its set of regexes. If a match is successful, the router will iterate over the configured evaluators configs and perform the configured action to submit the URL payload to the evaluator interface (either SNS topic or DAG submission). In this case, the router sees that the action is `submit_to_sns_topic` and thus publishes the URL payload (and the regular expression captured groups as `payload_info`) to the SNS topic (`topic_arn`) configured in the action's parameters. In addition to the payload URL and the payload info, the router also includes the `on_success` parameters configured for the action. This will propagate pertinent info to the underlying evaluator code which would be used if evaluation is successful. In this case, if the evaulator successfully evaluates that everything is ready for this input file, it can proceed to submit a DAG run for the `submit_nisar_tlm_ingest` DAG in the underlying SPS.
 
-Let's consider another example but this time the configured action is to submit a DAG run instead of publishing to an evaluator's SNS topic: 
+Let's consider another example but this time the configured action is to submit a DAG run instead of publishing to an evaluator's SNS topic:
 ```
 initiator_config:
   name: minimal config example
@@ -171,11 +171,13 @@ In this case, the router sees that the action is `submit_dag_by_id` and thus mak
 * [Quick Start](#quick-start)
   * [Setting Up the End-to-End Demo](#setting-up-the-end-to-end-demo)
     * [Deploying the Inititator](#deploying-the-initiator)
-    * [Deploying an Example Evaluator](#deploying-an-example-evaluator-sns-topic-sqs-queue-lambda)
+    * [Deploying Example Evaluators](#deploying-example-evaluators-sns-topic-sqs-queue-lambda)
     * [Deploying an S3 Event Notification Trigger](#deploying-an-s3-event-notification-trigger)
     * [Verify End-to-End Functionality (part 1)](#verify-end-to-end-functionality-part-1)
     * [Deploying an EventBridge Scheduler Trigger](#deploying-an-eventbridge-scheduler-trigger)
     * [Verify End-to-End Functionality (part 2)](#verify-end-to-end-functionality-part-2)
+    * [Deploying an EventBridge Scheduler Trigger for Periodic CMR Queries](#deploying-an-eventbridge-scheduler-trigger-for-periodic-cmr-queries)
+    * [Verify End-to-End Functionality (part 3)](#verify-end-to-end-functionality-part-3)
     * [Tear Down](#tear-down)
   * [Setup Instructions for Development](#setup-instructions-for-development)
   * [Build Instructions](#build-instructions)
@@ -212,13 +214,14 @@ This guide provides a quick way to get started with our project. Please see our
    ```
    cd unity-initiator/terraform-unity/initiator/
    ```
-1. Copy a sample router configuration YAML file to use for deployment and update the AWS region and AWS account ID to match your AWS environment. We will be using the NISAR TLM test case for this demo so we also rename the SNS topic ARN for it accordingly:
+1. Copy a sample router configuration YAML file to use for deployment and update the AWS region and AWS account ID to match your AWS environment. We will be using the NISAR TLM and AIRS RetStd test cases for this demo so we also rename the SNS topic ARNs for them accordingly:
    ```
    cp ../../tests/resources/test_router.yaml .
    export AWS_ACCOUNT_ID=$(aws sts get-caller-identity --output text | awk '{print $1}')
    export AWS_REGION=$(aws configure get region)
    sed -i "s/hilo-hawaii-1/${AWS_REGION}/g" test_router.yaml
    sed -i "s/123456789012:eval_nisar_ingest/${AWS_ACCOUNT_ID}:uod-dev-eval_nisar_ingest-evaluator_topic/g" test_router.yaml
+   sed -i "s/123456789012:eval_airs_ingest/${AWS_ACCOUNT_ID}:uod-dev-eval_airs_ingest-evaluator_topic/g" test_router.yaml
    ```
 1. You will need an S3 bucket for terraform to stage the router Lambda zip file during deployment. Create one or reuse an existing one and set an environment variable for it:
    ```
@@ -247,10 +250,19 @@ This guide provides a quick way to get started with our project. Please see our
    ```
    **Take note of the `initiator_topic_arn` that is output by terraform. It will be used when setting up any triggers.**
 
-#### Deploying an Example Evaluator (SNS topic->SQS queue->Lambda)
-1. Change directory to the location of the sns_sqs_lambda evaluator terraform:
+#### Deploying Example Evaluators (SNS topic->SQS queue->Lambda)
+##### Evaluator Deployment for NISAR TLM (via staged data to the ISL)
+1. Change directory to the location of the evaluators terraform:
    ```
-   cd ../evaluators/sns_sqs_lambda/
+   cd ../evaluators
+   ```
+1. Make a copy of the `sns_sqs_lambda` directory for the NISAR TLM evaluator:
+   ```
+   cp -rp sns_sqs_lambda sns_sqs_lambda-nisar_tlm
+   ```
+1. Change directory into the NISAR TLM evaluator terraform:
+   ```
+   cd sns_sqs_lambda-nisar_tlm/
    ```
 1. Set the name of the evaluator to our NISAR example:
    ```
@@ -270,8 +282,41 @@ This guide provides a quick way to get started with our project. Please see our
      --var evaluator_name=${EVALUATOR_NAME} \
      -auto-approve
    ```
-   **Take note of the `evaluator_topic_arn` that is output by terraform. It should match the topic ARN in the test_router.yaml file you used during the initiator deployment. If they match then the router Lambda is now able to submit payloads to this evaluator SNS topic.**
-   
+   **Take note of the `evaluator_topic_arn` that is output by terraform. It should match the respective topic ARN in the test_router.yaml file you used during the initiator deployment. If they match then the router Lambda is now able to submit payloads to this evaluator SNS topic.**
+
+##### Evaluator Deployment for AIRS RetStd (via scheduled CMR query)
+1. Change directory to the location of the evaluators terraform:
+   ```
+   cd ..
+   ```
+1. Make a copy of the `sns_sqs_lambda` directory for the AIRS RetStd evaluator:
+   ```
+   cp -rp sns_sqs_lambda sns_sqs_lambda-airs_retstd
+   ```
+1. Change directory into the AIRS RetStd evaluator terraform:
+   ```
+   cd sns_sqs_lambda-airs_retstd/
+   ```
+1. Set the name of the evaluator to our AIRS example:
+   ```
+   export EVALUATOR_NAME=eval_airs_ingest
+   ```
+1. Note the implementation of the evaluator code. It currently doesn't do any real evaluation but simply returns that evaluation was successful:
+   ```
+   cat data.tf
+   ```
+1. Initialize terraform:
+   ```
+   terraform init
+   ```
+1. Run terraform apply:
+   ```
+   terraform apply \
+     --var evaluator_name=${EVALUATOR_NAME} \
+     -auto-approve
+   ```
+   **Take note of the `evaluator_topic_arn` that is output by terraform. It should match the respective topic ARN in the test_router.yaml file you used during the initiator deployment. If they match then the router Lambda is now able to submit payloads to this evaluator SNS topic.**
+
 #### Deploying an S3 Event Notification Trigger
 1. Change directory to the location of the s3_bucket_notification trigger terraform:
    ```
@@ -341,9 +386,50 @@ This guide provides a quick way to get started with our project. Please see our
 1. The deployed EventBridge scheduler runs the trigger Lambda function with schedule expression of `rate(1 minute)`. After a minute, verify that the `eval_nisar_ingest` evaluator Lambda function was called successfully for each of those scheduled invocations by looking at its CloudWatch logs for entries similar to this:
    ![eval_log_2](https://github.com/unity-sds/unity-initiator/assets/387300/cae82e10-a736-43b7-8957-790fc29b5fea)
 
+#### Deploying an EventBridge Scheduler Trigger for Periodic CMR Queries
+1. Change directory to the location of the s3_bucket_notification trigger terraform:
+   ```
+   cd ../cmr_query/
+   ```
+1. Note the implementation of the trigger lambda code. It will query CMR for granules for a particular collection within a timeframe, query its dynamodb table if they already exist, and if not, submit them as payload URLs to the initiator SNS topic and save them into the dynamodb table:
+   ```
+   cat lambda_handler.py
+   ```
+1. Set the CMR provider ID for the AIRS RetStd collection:
+   ```
+   export PROVIDER_ID=GES_DISC
+   ```
+1. Set the CMR concept ID for the AIRS RetStd collection:
+   ```
+   export CONCEPT_ID=C1701805619-GES_DISC
+   ```
+1. Set the amount of seconds to look back from the current epoch for granules in the collection. For example, we will set this value to 2 days (172800 seconds) so that when the CMR query lambda kicks off, it will query for all AIRS RetStd granules using a temporal search of `now - 172800 seconds` to `now`:
+   ```
+   export SECONDS_BACK=172800
+   ```
+1. Initialize terraform:
+   ```
+   terraform init
+   ```
+1. Run terraform apply. Note the DEPLOYMENT_NAME, CODE_BUCKET and INITIATOR_TOPIC_ARN environment variables should have been set in the previous steps. If not set them again:
+   ```
+   terraform apply \
+     --var deployment_name=${DEPLOYMENT_NAME} \
+     --var code_bucket=${CODE_BUCKET} \
+     --var initiator_topic_arn=${INITIATOR_TOPIC_ARN} \
+     --var provider_id=${PROVIDER_ID} \
+     --var concept_id=${CONCEPT_ID} \
+     --var seconds_back=${SECONDS_BACK} \
+     -auto-approve
+   ```
+
+#### Verify End-to-End Functionality (part 3)
+1. The deployed EventBridge scheduler runs the trigger CMR query Lambda function with schedule expression of `rate(1 minute)`. After a minute, verify that the `eval_airs_ingest` evaluator Lambda function was called successfully for each of those scheduled invocations by looking at its CloudWatch logs for entries similar to this:
+   ![eval_log_3](https://github.com/user-attachments/assets/54b26349-91b2-4958-9082-47613da6c675)
+
 #### Tear Down
 1. Simply go back into each of the terraform directories for which `terraform apply` was run and run `terraform destroy`.
-   
+
 ### Setup Instructions for Development
 
 1. Clone repo:

scripts/build_cmr_query_lambda_package.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+BASE_PATH=$(dirname "${BASH_SOURCE}")
+BASE_PATH=$(cd "${BASE_PATH}/.."; pwd)
+DIST_DIR=${BASE_PATH}/dist
+PKG_DIR=${DIST_DIR}/lambda_packages
+CMR_QUERY_DIR=${BASE_PATH}/terraform-unity/triggers/cmr_query
+
+set -ex
+
+rm -rf $DIST_DIR
+pip install hatch
+hatch clean
+hatch build
+VERSION=$(hatch run python -c 'from importlib.metadata import version; print(version("unity_initiator"))')
+echo "{\"version\": \"$VERSION\"}" > ${DIST_DIR}/version.json
+mkdir -p $PKG_DIR
+pip install -t $PKG_DIR ${DIST_DIR}/unity_initiator-*.whl
+pip install -t $PKG_DIR python_cmr
+cp ${CMR_QUERY_DIR}/lambda_handler.py $PKG_DIR/
+cd $PKG_DIR
+zip -rq ${DIST_DIR}/cmr_query-${VERSION}-lambda.zip .

scripts/build_lambda_package.sh

@@ -3,7 +3,6 @@ BASE_PATH=$(dirname "${BASH_SOURCE}")
 BASE_PATH=$(cd "${BASE_PATH}/.."; pwd)
 DIST_DIR=${BASE_PATH}/dist
 PKG_DIR=${DIST_DIR}/lambda_packages
-TEST_DIR=${BASE_PATH}/tests
 
 set -ex
 

src/unity_initiator/cloud/lambda_handler.py

@@ -57,25 +57,25 @@ def lambda_handler_initiator(event, context):
         # TODO: Find cleaner way of parsing payloads from a variety of sources (S3 event notification, EventBridge->Lambda).
         # For now we use brittle assumptions on the payload structure for each of the supported sources.
 
-        # skip S3 test event
-        if message.get("Event", None) == "s3:TestEvent":
-            logger.info("Skipped s3:TestEvent")
-            continue
-
-        # payload comes from S3 notification
-        if "Records" in message:
-            for rec in message["Records"]:
-                s3_info = rec["s3"]
-                payloads.append(
-                    {
-                        "payload": f"s3://{s3_info['bucket']['name']}/{s3_info['object']['key']}"
-                    }
-                )
-        # payload comes from EventBridge scheduled task
+        if isinstance(message, dict):
+            # skip S3 test event
+            if message.get("Event", None) == "s3:TestEvent":
+                logger.info("Skipped s3:TestEvent")
+                continue
+
+            # payload comes from S3 notification
+            if "Records" in message:
+                for rec in message["Records"]:
+                    s3_info = rec["s3"]
+                    payloads.append(
+                        {
+                            "payload": f"s3://{s3_info['bucket']['name']}/{s3_info['object']['key']}"
+                        }
+                    )
+        elif isinstance(message, list):
+            # payload comes from EventBridge scheduled task
+            payloads.extend(message)
         else:
-            if isinstance(message, list):
-                payloads.extend(message)
-            else:
-                payloads.append(message)
+            payloads.append(message)
 
     return lambda_handler_multiple_payloads(payloads, context)

terraform-unity/evaluators/sns_sqs_lambda/README.md

@@ -26,6 +26,7 @@ No modules.
 
 | Name | Type |
 |------|------|
+| [aws_cloudwatch_log_group.evaluator_lambda_log_group](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/cloudwatch_log_group) | resource |
 | [aws_iam_policy.evaluator_lambda_policy](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_policy) | resource |
 | [aws_iam_role.evaluator_lambda_iam_role](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role) | resource |
 | [aws_iam_role_policy_attachment.lambda_base_policy_attachment](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role_policy_attachment) | resource |

terraform-unity/evaluators/sns_sqs_lambda/locals.tf

@@ -1,4 +1,5 @@
 locals {
+  function_name = "${var.project}-${var.venue}-${var.evaluator_name}-evaluator"
   tags = {
     Venue       = "dev"
     ServiceArea = "cs"

terraform-unity/evaluators/sns_sqs_lambda/main.tf

@@ -8,7 +8,7 @@ resource "aws_lambda_function" "evaluator_lambda" {
     ]
   }
 
-  function_name    = "${var.project}-${var.venue}-${var.evaluator_name}-evaluator"
+  function_name    = local.function_name
   role             = aws_iam_role.evaluator_lambda_iam_role.arn
   handler          = "lambda_function.lambda_handler"
   runtime          = "python3.11"
@@ -18,8 +18,13 @@ resource "aws_lambda_function" "evaluator_lambda" {
   tags             = local.tags
 }
 
+resource "aws_cloudwatch_log_group" "evaluator_lambda_log_group" {
+  name              = "/aws/lambda/${local.function_name}"
+  retention_in_days = 14
+}
+
 resource "aws_iam_role" "evaluator_lambda_iam_role" {
-  name = "${var.project}-${var.venue}-${var.evaluator_name}-evaluator_lambda_iam_role"
+  name = "${local.function_name}_lambda_iam_role"
 
   assume_role_policy = jsonencode({
     Version = "2012-10-17",
@@ -38,7 +43,7 @@ resource "aws_iam_role" "evaluator_lambda_iam_role" {
 }
 
 resource "aws_iam_policy" "evaluator_lambda_policy" {
-  name        = "${var.project}-${var.venue}-${var.evaluator_name}-evaluator_lambda_policy"
+  name        = "${local.function_name}_lambda_policy"
   description = "A policy for the evaluator lambda function to access S3 and SQS"
 
   policy = jsonencode({
@@ -89,7 +94,7 @@ resource "aws_ssm_parameter" "evaluator_lambda_function_name" {
 
 
 resource "aws_sqs_queue" "evaluator_dead_letter_queue" {
-  name                       = "${var.project}-${var.venue}-${var.evaluator_name}-evaluator_dead_letter_queue"
+  name                       = "${local.function_name}_dead_letter_queue"
   delay_seconds              = 0
   max_message_size           = 2048
   message_retention_seconds  = 1209600
@@ -99,7 +104,7 @@ resource "aws_sqs_queue" "evaluator_dead_letter_queue" {
 }
 
 resource "aws_sqs_queue" "evaluator_queue" {
-  name                       = "${var.project}-${var.venue}-${var.evaluator_name}-evaluator_queue"
+  name                       = "${local.function_name}_queue"
   delay_seconds              = 0
   max_message_size           = 2048
   message_retention_seconds  = 1209600
@@ -113,7 +118,7 @@ resource "aws_sqs_queue" "evaluator_queue" {
 }
 
 resource "aws_sns_topic" "evaluator_topic" {
-  name = "${var.project}-${var.venue}-${var.evaluator_name}-evaluator_topic"
+  name = "${local.function_name}_topic"
 }
 
 resource "aws_sns_topic_policy" "evaluator_topic_policy" {

terraform-unity/initiator/README.md

@@ -26,6 +26,7 @@ No modules.
 
 | Name | Type |
 |------|------|
+| [aws_cloudwatch_log_group.initiator_lambda_log_group](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/cloudwatch_log_group) | resource |
 | [aws_iam_policy.initiator_lambda_policy](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_policy) | resource |
 | [aws_iam_role.initiator_lambda_iam_role](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role) | resource |
 | [aws_iam_role_policy_attachment.lambda_base_policy_attachment](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/iam_role_policy_attachment) | resource |

terraform-unity/initiator/locals.tf

@@ -1,4 +1,5 @@
 locals {
+  function_name = "${var.project}-${var.venue}-${var.deployment_name}-inititator"
   tags = {
     Venue       = "dev"
     ServiceArea = "cs"