#!/bin/sh # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # Permission is hereby granted, free of charge, to any person obtaining a copy of # this software and associated documentation files (the "Software"), to deal in # the Software without restriction, including without limitation the rights to # use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of # the Software, and to permit persons to whom the Software is furnished to do so. # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS # FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR # COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER # IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. if ! command -v aws &> /dev/null then echo "aws cli not installed" exit 1 fi if ! command -v jq &> /dev/null then echo "jq is not installed" exit 1 fi if ! command -v head &> /dev/null then echo "head is not installed" exit 1 fi if ! command -v awk &> /dev/null then echo "awk is not installed" exit 1 fi if ! command -v sleep &> /dev/null then echo "sleep is not installed" exit 1 fi REGION=$(aws configure get default.region) if [ -z "$REGION" ]; then echo "no default region, using us-east-1" REGION='us-east-1' fi VERSION=$(aws --version 2>&1) if [ "${VERSION:0:10}" != 'aws-cli/2.' ]; then echo "please update to awscli version 2" echo "https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html" exit 1 fi # test if aws cli has mwaa, if not prompt for upgrade aws mwaa list-environments --region $REGION > /dev/null 2>&1 RESULT=$? if [ $RESULT -ne 0 ]; then echo "please update the aws cli, mwaa not found" exit 1 fi echo "Number of arguments: $#" echo "args: $@" for key in "$@" do case $key in --envname=*) NAME="${key#*=}" if [ -z "$NAME" ];then echo "--envname is empty, requires a string value" exit 1 fi echo "NAME = $NAME" shift ;; --dag=*) DAG="${key#*=}" if [ -z "$DAG" ];then echo "--dag is empty, requires a string value" exit 1 fi echo "DAG = $DAG" shift ;; --command=*) COMMAND="${key#*=}" if [ -z "$COMMAND" ];then echo "--command is empty, requires a string value" exit 1 fi COMMAND="{\"command\":\"${key#*=}\"}" echo "COMMAND = $COMMAND" shift ;; --region=*) REGION="${key#*=}" if [ -z "$REGION" ];then REGION=$(aws configure get default.region) echo "no region, using default region $REGION" fi shift ;; *) echo "unknown option" exit 1 ;; esac done # validate required params are there before running curl command if [ -z "$NAME" ];then echo "--envname is required, please rerun with --envname=SOME_VALUE" exit 1 fi if [ -z "$DAG" ];then echo "--dag is required, please rerun with --dag=SOME_VALUE" exit 1 fi if [ -z "$COMMAND" ];then echo "--command is required, please rerun with --command=SOME_VALUE" exit 1 fi echo "region = $REGION" STD_OUTPUT="" STD_ERROR="" CLI_JSON=$(aws mwaa create-cli-token --name $NAME --region $REGION) \ && CLI_TOKEN=$(echo $CLI_JSON | jq -r '.CliToken') \ && WEB_SERVER_HOSTNAME=$(echo $CLI_JSON | jq -r '.WebServerHostname') \ && CLI_RESULTS=$(curl -s --request POST "https://$WEB_SERVER_HOSTNAME/aws_mwaa/cli" \ --header "Authorization: Bearer $CLI_TOKEN" \ --header "Content-Type: text/plain" \ --data-raw "trigger_dag $DAG --conf '$COMMAND'") \ && echo -n "Sent request to airflow using curl, Response body:" \ && STD_OUTPUT=$(echo $CLI_RESULTS | jq -r '.stdout' | base64 --decode) \ && echo $STD_OUTPUT \ && echo -n "Errors:" \ && STD_ERROR=$(echo $CLI_RESULTS | jq -r '.stderr' | base64 --decode) \ && echo $STD_ERROR RESULT=$? if [ $RESULT -ne 0 ]; then exit 1 fi # query for specific log stream to get exact logs LOG_STREAM_PREFIX=$DAG/bash_command/$(echo $STD_OUTPUT | head -n 1 | awk -F"[][]" '{print $2}' | awk -F"\ |," '{gsub(":","_",$2); print $1"T"$2}') # query for dag status to make sure its finished and loop until if finishes echo "waiting for run to finish..." while [ true ] do STD_OUTPUT="running" CLI_JSON=$(aws mwaa create-cli-token --name $NAME) CLI_TOKEN=$(echo $CLI_JSON | jq -r '.CliToken') \ && WEB_SERVER_HOSTNAME=$(echo $CLI_JSON | jq -r '.WebServerHostname') \ && CLI_RESULTS=$(curl -s "https://$WEB_SERVER_HOSTNAME/aws_mwaa/cli" \ --header "Authorization: Bearer $CLI_TOKEN" \ --header "Content-Type: application/json" \ --data-raw "list_dag_runs --state running $DAG ") \ && echo -n "Dag status: " \ && STD_OUTPUT=$(echo $CLI_RESULTS | jq -r '.stdout' | base64 --decode) \ && echo $STD_OUTPUT | awk -F"DAG RUNS" '{print $2}' | awk -F"|" '{print $8}' \ && echo -n "Errors:" \ && STD_ERROR=$(echo $CLI_RESULTS | jq -r '.stderr' | base64 --decode) \ && echo $STD_ERROR if [ -z "$(echo $STD_OUTPUT | grep "running")" ]; then echo "dag has finished running, printing logs:" break fi echo "waiting for 10 seconds..." sleep 10 done # sleep for another 3 seconds just to give the logs a chance to get to the stream in time to query them sleep 3 aws logs filter-log-events --log-group-name "airflow-$NAME-Task" --log-stream-name-prefix "$LOG_STREAM_PREFIX" --region $REGION --output text --query 'sort_by(events, ×tamp)[*].[message]'