Airflow Integrated with Slack Actions

[emincanoguz11]

Hi,
I tried to active clear button. But it is not works 😞 Also I give this library to the top of source code but this is not recommended for Airflow because of 30 second rules. Also I tried added my on failure callback alarm it is not works 😞 Can we manage actions for button in Airflow?
Any guess to helping me? Maybe this feature is get PR I don’t know

[emincanoguz11]

Hi everyone,

I did it :) Thanks for @eladkal and @Taragolis for kindly supporting

When I try integrated actions in Airflow and I see that result success 🙂
Basically this integration is very complicated and the main result is slack is not accept to the webhook operators with actions. You must add client to the giving response to the slack.
Following code is the example of the see result

def ingestionErrorAlert(context):
    slack_webhook_token = BaseHook.get_connection(SLACK_CONN_ID).password
    from slack_bolt import App
    from slack_sdk import WebClient
    from slack_sdk.errors import SlackApiError
    from slack_bolt.adapter.socket_mode import SocketModeHandler
    app = App(verification_token=VERIFICATION_TOKEN, signing_secret=SIGNING_SECRET,
              token=BOT_TOKEN)
    client = WebClient(token=BOT_TOKEN)

    try:
        response = client.chat_postMessage(
            channel="#qos-airflow-alert-deneme",
            text='Example',
            blocks=[
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": ":red_circle: Task Failed. <!here>"
                    }
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": "*Task Id:*\n {} \n*Dag Id:*\n {} \n *Execution Time:*\n {} \n*Log URL:* http://airflow-qos-k8s-r1.hepsiburada.com/dags/appended-listings/grid\n`".format(
                            context.get('task_instance').task_id, context.get('task_instance').dag_id,
                            context.get('execution_date'))
                    }
                },
                {
                    "type": "actions",
                    "elements": [
                        {
                            "type": "button",
                            "action_id": "clear",
                            "text": {
                                "type": "plain_text",
                                "emoji": True,
                                "text": "Clear"
                            },
                            "style": "primary",
                            "value": "click_me_123"
                        }
                    ]
                }
            ]
        )
        print(response)
    except SlackApiError as e:
        print("Error sending message: {}".format(e))

    @app.action("clear")
    def button_press(ack, respond, action):
        ack()


        response = client.chat_postMessage(
            channel="#qos-airflow-alert-deneme",
            text="Button pressed!"
        )
        print(response)
        print(response.data)

    SocketModeHandler(app,
                      APP_TOKEN).start()

As you clearly see that this code is so messy 🙂 Because of the Airflow is not supported to the adding library firstly (30 second time covered to the this problem)
Also you must add bot in using channel as a screenshot because of when you don’t add app slack is not see and you will see channel is not found error.

P.S Also you add any jobs in actions function. I tried that clear the dag and I success 🙂 This is example for how to implement this

P.S Maybe I PRing to the implement client type for using Airflow (for example I will implement AirflowSlackClientOperator etc) Basically this is same code blocks for many times but this is changed for blocks and actions.

[Taragolis]

P.S Maybe I PRing to the implement client type for using Airflow (for example I will implement AirflowSlackClientOperator etc) Basically this is same code blocks for many times but this is changed for blocks and actions.

Airflow support both Slack API (connection ref) and Slack Incoming Webhook (connection ref) integration.

However main idea for this integration call within the tasks or callbacks (in Airflow 2.6+ also specific helpers would added) and not create some web application.

If you need create your web application which accept some incoming requests then you could:

1. Create it as standalone service and deploy it separately
2. You could (no idea is it possible or not) try also try to achieve it with Plugin appbuilder_views functionality

[Taragolis]

Also Airflow have REST API access so you don't need to run command as shell applications. There is also a python client for communicate with REST API: airflow-client-python

[emincanoguz11]

I understand that but if I manage like event handler how I integrated with Airflow? When I write this dag, I think about what is the results of adding actions in Dag? I think this is successful because of this is managed only in dag. (Also my second plan is manage slack actions in event handler and airflow send this event in event handler and I managed in event handler code)

But I understand that Airflow isn't not service orchestration. Means that Airflow is only running to the pipelines.

Thanks for your support @Taragolis again :)

[emincanoguz11]

Also Airflow have REST API access so you don't need to run command as shell applications. There is also a python client for communicate with REST API: airflow-client-python

Yeah I know that but CLI is very good solution for this requirements. Because I don't love Airflow's REST API. Airflow Rest API is complicated than other REST API. I managed with CLI. Thanks for advice :)