-
Notifications
You must be signed in to change notification settings - Fork 90
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add pubsub and slack example operators
- Loading branch information
Showing
7 changed files
with
137 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| from airflow import DAG | ||
| from datetime import datetime, timedelta | ||
| from utils.tags import Tag | ||
|
|
||
| from airflow.providers.google.cloud.operators.pubsub import PubSubPublishMessageOperator | ||
|
|
||
| default_args = { | ||
| "owner": "[email protected]", | ||
| "email": ["[email protected]"], | ||
| "depends_on_past": False, | ||
| "start_date": datetime(2022, 3, 23), | ||
| "email_on_failure": True, | ||
| "email_on_retry": True, | ||
| "retries": 2, | ||
| "retry_delay": timedelta(minutes=30), | ||
| } | ||
|
|
||
| m1 = {'data': b'Hello, World!', | ||
| 'attributes': {'type': 'greeting'} | ||
| } | ||
| m2 = {'data': b'Knock, knock'} | ||
| m3 = {'attributes': {'foo': ''}} | ||
|
|
||
| tags = [Tag.ImpactTier.tier_3] | ||
| dag = DAG("example_pubsub", default_args=default_args, schedule_interval="@daily", tags=tags,) | ||
|
|
||
| publish_task = PubSubPublishMessageOperator( | ||
| task_id="publish_to_pubsub", | ||
| project_id="moz-fx-data-airflow-prod-88e0", | ||
| topic="airflow-glam-triggers", | ||
| gcp_conn_id="google_cloud_airflow_pubsub", | ||
| messages=[m1, m2, m3], | ||
| dag=dag, | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| from airflow import DAG | ||
| from airflow.models import Variable | ||
| from datetime import datetime, timedelta | ||
| from utils.tags import Tag | ||
|
|
||
| from airflow.providers.slack.operators.slack import SlackAPIPostOperator | ||
| from airflow.operators.bash import BashOperator | ||
|
|
||
| from utils.slack import if_task_fails_alert_slack | ||
|
|
||
| """ | ||
| If getting "channel_not_found" errors, you need to open the slack channel settings, navigate to Integrations, | ||
| and add "Airflow-bot" to the Apps section. | ||
| """ | ||
| tags = [Tag.ImpactTier.tier_3] | ||
|
|
||
| """ | ||
| default_args_1 = { | ||
| "owner": "[email protected]", | ||
| "email": ["[email protected]"], | ||
| "depends_on_past": False, | ||
| "start_date": datetime(2022, 3, 23), | ||
| "email_on_failure": True, | ||
| "email_on_retry": True, | ||
| "retries": 2, | ||
| "retry_delay": timedelta(minutes=30), | ||
| } | ||
| dag1 = DAG("example_slack", default_args=default_args_1, schedule_interval="@daily", tags=tags,) | ||
| # The following example shows how to simply post to a slack channel | ||
| simple_slack_example = SlackAPIPostOperator( | ||
| task_id="post_hello", | ||
| token=Variable.get("slack_secret_token"), | ||
| text="hello world!", | ||
| channel="#airflow-alerts", | ||
| dag=dag1, | ||
| ) | ||
| """ | ||
|
|
||
| # This example shows how to configure a dag's default args callback so it alerts slack on failures using an | ||
| # imported utils method | ||
| default_args_2 = { | ||
| "owner": "[email protected]", | ||
| "email": ["[email protected]"], | ||
| "depends_on_past": False, | ||
| "start_date": datetime(2022, 3, 23), | ||
| "email_on_failure": True, | ||
| "email_on_retry": True, | ||
| "retries": 0, | ||
| "retry_delay": timedelta(minutes=30), | ||
| # NOTE: on_failure_callback doesn't trigger until all retries are exhausted | ||
| "on_failure_callback": if_task_fails_alert_slack, | ||
| } | ||
|
|
||
| dag2 = DAG("example_slack", default_args=default_args_2, schedule_interval="@daily", tags=tags,) | ||
|
|
||
| task_with_failed_slack_alerts = BashOperator( | ||
| task_id='fail_task', | ||
| bash_command='exit 1', | ||
| dag=dag2) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| from airflow.models import Variable | ||
| from airflow.providers.slack.operators.slack import SlackAPIPostOperator | ||
|
|
||
| SLACK_CHANNEL = "#airflow-alerts" | ||
|
|
||
| def if_task_fails_alert_slack(context): | ||
| failed_alert = SlackAPIPostOperator( | ||
| task_id='slack_failed', | ||
| channel=SLACK_CHANNEL, | ||
| token=Variable.get("slack_secret_token"), | ||
| text=""" | ||
| :red_circle: Task Failed. | ||
| *Task*: {task} | ||
| *Dag*: {dag} | ||
| *Date*: {ds} | ||
| """.format( | ||
| task=context.get('task_instance').task_id, | ||
| dag=context.get('task_instance').dag_id, | ||
| ds=context.get('ds') | ||
| ) | ||
| ) | ||
| return failed_alert.execute(context=context) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters