apache / apache/airflow

@hookimpl on_dag_run_running, on_dag_run_success, on_dag_run_failed do not find Connections and Variables

Open
#39,646 3 comments 0 reactions 0 assignees View on GitHub
_eol_affected_version:2.8 area:Listeners kind:bug
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 10h
Merged PRs (30d)
483

Description

### Apache Airflow version

Other Airflow 2 version (please specify below)

### If "Other Airflow 2 version" selected, which one?

2.8.1

### What happened?

I wrote a listener plugin and added it to the plugins directory. The listener includes the methods:
- on_task_instance_running
- on_task_instance_success
- on_task_instance_failed
- on_dag_run_running
- on_dag_run_success
- on_dag_run_failed.
When trying to extract values from Variables and Connections in the on_dag_run_running, on_dag_run_success, on_dag_run_failed methods, an error occurs already when running dag run:
```
Traceback (most recent call last):

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 52, in _run_scheduler_job

run_job(job=job_runner.job, execute_callable=job_runner._execute)

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/session.py", line 79, in wrapper

return func(*args, session=session, **kwargs)

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/job.py", line 393, in run_job

return execute_job(job, execute_callable=execute_callable)

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/job.py", line 422, in execute_job

ret = execute_callable()

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job_runner.py", line 855, in _execute

self._run_scheduler_loop()

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job_runner.py", line 987, in _run_scheduler_loop

num_queued_tis = self._do_scheduling(session)

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job_runner.py", line 1063, in _do_scheduling

self._start_queued_dagruns(session)

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/scheduler_job_runner.py", line 1404, in _start_queued_dagruns

dag_run.notify_dagrun_state_changed()

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/dagrun.py", line 862, in notify_dagrun_state_changed

get_listener_manager().hook.on_dag_run_running(dag_run=self, msg=msg)

File "/home/airflow/.local/lib/python3.9/site-packages/pluggy/_hooks.py", line 493, in __call__

return self._hookexec(self.name, self._hookimpls, kwargs, firstresult)

File "/home/airflow/.local/lib/python3.9/site-packages/pluggy/_manager.py", line 115, in _hookexec

return self._inner_hookexec(hook_name, methods, kwargs, firstresult)

File "/home/airflow/.local/lib/python3.9/site-packages/pluggy/_callers.py", line 113, in _multicall

raise exception.with_traceback(exception.__traceback__)

File "/home/airflow/.local/lib/python3.9/site-packages/pluggy/_callers.py", line 77, in _multicall

res = hook_impl.function(*args)

File "/opt/airflow/plugins/metadata/airflow_metadata.py", line 206, in on_dag_run_running

my_connection = BaseHook.get_connection("my_connection")

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/hooks/base.py", line 82, in get_connection

conn = Connection.get_connection_from_secrets(conn_id)

File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/connection.py", line 479, in get_connection_from_secrets

raise AirflowNotFoundException(f"The conn_id `{conn_id}` isn't defined")

airflow.exceptions.AirflowNotFoundException: The conn_id `my_connection` isn't defined
```

Although the connection is explicitly defined in Connections and everything works in the case of listening to tasks.

How can I fix it so that if I listen to Dog Run, everything works too?

### What you think should happen instead?

I expect the values to be pulled up from Connection and Variable when listening to DagRun.

### How to reproduce

Add to Connection "my_connection" and add to Variable "environment"

```
from airflow.listeners import hookimpl
from airflow.models.taskinstance import TaskInstance
from airflow.hooks.base import BaseHook
from airflow.models import Variable
from airflow.models.dagrun import DagRun
from airflow.plugins_manager import AirflowPlugin

class AirflowListener:
@hookimpl
def on_task_instance_running(self, task_instance: TaskInstance) -> None:
my_connection = BaseHook.get_connection("my_connection")
env = Variable.get("environment")

@hookimpl
def on_task_instance_success(self, task_instance: TaskInstance) -> None:
my_connection = BaseHook.get_connection("my_connection")
env = Variable.get("environment")

@hookimpl
def on_task_instance_failed(self, task_instance: TaskInstance) -> None:
my_connection = BaseHook.get_connection("my_connection")
env = Variable.get("environment")

@hookimpl
def on_dag_run_running(self, dag_run: DagRun):
my_connection = BaseHook.get_connection("my_connection")
env = Variable.get("environment")

@hookimpl
def on_dag_run_success(self, dag_run: DagRun):
my_connection = BaseHook.get_connection("my_connection")
env = Variable.get("environment")

@hookimpl
def on_dag_run_failed(self, dag_run: DagRun):
my_connection = BaseHook.get_connection("my_connection")
env = Variable.get("environment")


class AirflowListenerPlugin(AirflowPlugin):
name = "AirflowListener"
listeners = [AirflowListener()]

```

### Operating System

macOS Sonoma 14.1.2

### Versions of Apache Airflow Providers

_No response_

### Deployment

Docker-Compose

### Deployment details

apache/airflow:2.8.1-python3.9
executor: CeleryExecutor

### Anything else?

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

Start by tracing the DAG-run listener call from airflow/models/dagrun.py, notify_dagrun_state_changed, through airflow/hooks/base.py and the connection lookup shown in the traceback. Compare that path with the task-instance listener path in the reproduction. Done means the three DAG-run hooks can retrieve the configured connection and variable, with coverage for the reported failure.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
backend
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.