Skip to content

Add Airflow version in the Slack report for master DAG run #521

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jul 14, 2022
Merged

Add Airflow version in the Slack report for master DAG run #521

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e507e85
Add Airflow version in the Slack report for master DAG run
pankajkoti Jul 14, 2022
87010f4
Include Airflow version as first line of the report
pankajkoti Jul 14, 2022
f9ca41f
Update .circleci/integration-tests/master_dag.py
pankajkoti Jul 14, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions .circleci/integration-tests/master_dag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import os
import time
from datetime import datetime
from typing import List
from typing import Any, List

from airflow import DAG
from airflow.contrib.operators.slack_webhook_operator import SlackWebhookOperator
Expand All @@ -19,7 +19,7 @@
SLACK_USERNAME = os.getenv("SLACK_USERNAME", "airflow_app")


def get_report(dag_run_ids: List[str]) -> None:
def get_report(dag_run_ids: List[str], **context: Any) -> None:
"""Fetch dags run details and generate report"""
with create_session() as session:
last_dags_runs: List[DagRun] = session.query(DagRun).filter(DagRun.run_id.in_(dag_run_ids)).all()
Expand All @@ -39,6 +39,10 @@ def get_report(dag_run_ids: List[str]) -> None:
task_message_str = f"{task_code} {ti.task_id} : {ti.state} \n"
message_list.append(task_message_str)

airflow_version = context["ti"].xcom_pull(task_ids="get_airflow_version")
airflow_version_message = f"\n\nAirflow version for the run is `{airflow_version}`"
message_list.append(airflow_version_message)

logging.info("%s", "".join(message_list))
# Send dag run report on Slack
try:
Expand Down Expand Up @@ -94,6 +98,10 @@ def prepare_dag_dependency(task_info, execution_time):
task_id="list_installed_pip_packages", bash_command="pip freeze"
)

get_airflow_version = BashOperator(
task_id="get_airflow_version", bash_command="airflow version", do_xcom_push=True
)

dag_run_ids = []
# AWS S3 and Redshift DAG
amazon_task_info = [
Expand Down Expand Up @@ -194,6 +202,7 @@ def prepare_dag_dependency(task_info, execution_time):
python_callable=get_report,
op_kwargs={"dag_run_ids": dag_run_ids},
trigger_rule="all_done",
provide_context=True,
)

end = DummyOperator(
Expand All @@ -203,6 +212,7 @@ def prepare_dag_dependency(task_info, execution_time):

start >> [
list_installed_pip_packages,
get_airflow_version,
amazon_trigger_tasks[0],
emr_trigger_tasks[0],
google_trigger_tasks[0],
Expand All @@ -218,6 +228,7 @@ def prepare_dag_dependency(task_info, execution_time):

last_task = [
list_installed_pip_packages,
get_airflow_version,
amazon_trigger_tasks[-1],
emr_trigger_tasks[-1],
google_trigger_tasks[-1],
Expand Down