feat: Update MWAATriggerDagsStage to pass json event to MWAA airflow dag

[devansh-gandhi]

Description

Currently, MWAATriggerDagStage does not pass the sfn event to the MWAA dag. This enhancement enables to pass runtime parameters/sfn event when triggering dag. The parameters can be accessed from the dag_run.conf.

Changes Made

Changes made to MWAATriggerDagStage:

1. Trigger dag lambda has an additional payload of the sfn event passed in body
2. airflow command updated to pass sfn event to MWAA dag using --conf

Related Issue

#428

[malachi-constant]

AWS CodeBuild CI Report

• CodeBuild project: DDKIntegrationTestsD8472609-5SCHWjldmWcf
• Commit ID: c5c6b59
• Result: SUCCEEDED
• Build Logs (available for 30 days)

Powered by github-codebuild-logs, available on the AWS Serverless Application Repository

[anmolsgandhi]

Looks Good, i think this would be useful in all the cases where the dags needs the input from sfn event for processing. @malachi-constant would this need any test coverage?

[malachi-constant]

In the case of an empty event is dags trigger DAG_ID --conf {} valid?

[devansh-gandhi]

Yes, in case of an empty event, dags trigger DAG_ID --conf {} is valid.

To test this, i created a sample dag the prints out the conf. (logger.info(context['dag_run'].conf))

Scenario 1: In case of a dynamic dag ids passed through dag_path, the event body is not expected to be empty as the sfn event has the dag_ids list. The test dag completed execution and printed the dag_ids list in the logs.

Scenario 2: To test a static dag id, i triggered the dag from the mwaa ui using the "Trigger DAG w/ Config" option. The configuration JSON (conf) was set to {}. The dag completed execution and printed {}