Static file import in DAG is not reloaded on the scheduler

[kobethuwis]

Using Airflow 2.5.1, deployed through the 1.8.0 helm chart. We're using lot's of external modules which are all located and imported through the /plugins folder. These are all simple Python modules, not AirflowPlugins.

A generic set of import in our DAGs looks something like this:

from airflow import DAG
from helpers.failure_functions import dag_failure_callback
from operators.pooled_python_operator import PooledPythonOperator
from variables import dag_names

When pushing the changes to the correct branch, the gitsync sidecar updates all modules & files and syncs them to the scheduler pod (I confirmed this by actually tailing the python files in question in the airflow-scheduler container). After several seconds/minutes, all of these changes come through in the UI.

EXCEPT for changes to our dag_names file: it is not a package, class or method, but a simple python file containing strings. Whenever we're updating this file, the changes are synced to the scheduler pod but aren't synced to the Airflow UI.

Currently we're fixing this by restarting the scheduler pod everytime we're adding a DAG to our list of DAG names.

Why is this happening? Are we importing the file in a bad manner? Is Airflow unable to package a static file? Or should we review our usage of external python modules and start using AirflowPlugin wrappers instead?

[potiuk]

EXCEPT for changes to our dag_names file: it is not a package, class or method, but a simple python file containing strings. Whenever we're updating this file, the changes are synced to the scheduler pod but aren't synced to the Airflow UI.

What exactly do you expect to be changed? Can you provide content of your dag_names, example of changes you made and screenshot of what you see "before" and "after" you see in UI? You missed that part in your explanation.

[kobethuwis]

In example, when I develop a new dag, I'll:

• Write a new python file and place it in /dags/new_dag.py
• Add it's name to the plugins/variables/dag_names.py file

MY_DAG = "My_DAG"
NEW_DAG = "New_DAG"

• Retrieve it's name in the dag using

from variables import dag_tags
dag = DAG(dag_names.NEW_DAG, ...)

When I deploy the new DAG, I receive an import error on dag_names.NEW_DAG. When I change something on non-static files in the /plugins folder, the changes are picked up as expected. Local exec into the pod shows that the git sync did pick up the commit and contains the updated dag_names file. The import error is resolved whenever we kill the scheduler pod.

[potiuk]

Why plugins.?

Can you put it in the "dag" - plugins are not supposed to be used for that. Follow https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/modules_management.html#modules-management as an example.

[kobethuwis]

We put our python modules in the /plugins folder since it's automatically added to the $PYTHONPATH. I could put it in a different folder and at it to the PYTHONPATH myself, but why would this fix the underlying problem?

[kobethuwis]

I got it to work: it was both a Python packaging & airflow issue. Had to add the following to the chart to force redeploy the scheduler once plugins we're changed:

extraEnv: |
  - name: AIRFLOW__WEBSERVER__RELOAD_ON_PLUGIN_CHANGE
    value: "True"

Next to that I needed to specify init.py files in the plugins folder and all of the subdirs. The fact that 'static' files we're not picked up was a symptom of bad python pckaging of the plugins dir, which Visual Studio was actually hinting at. as well. (module not found and no way to resolve it)

[potiuk]

Yes. This is a potential problem that we even explicitly spelled out as a separate chapter in the "modules management" documentation:

https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/modules_management.html#add-init-py-in-package-folders

When you create folders you should add init.py file as empty files in your folders. While in Python 3 there is a concept of implicit namespaces where you do not have to add those files to folder, Airflow expects that the files are added to all packages you added.

[kobethuwis]

Which helped me a lot indeed, would have never found this without the doc!

[potiuk]

Glad to hear it :). Usually we got the feedback that thigs are "missing" in our docs, so getting a positive one that our docs have helped is a nice one for a change.

[kobethuwis]

@potiuk I'm again experiencing issues with this; especially on a folder called 'variables'.

I just read this:

It is recommended that you always put your DAGs/common files in a subpackage which is unique to your deployment (my_company in the example below). It is far too easy to use generic names for the folders that will clash with other packages already present in the system. For example if you create airflow/operators subfolder it will not be accessible because Airflow already has a package named airflow.operators and it will look there when importing from airflow.operators.

Is there a quick way to check if the 'variables' module is already defined somewhere in airflow?

I'm clueless, I've tested everything else:

x my modules contain the init.py files

x I've configured my chart correctly to reload the plugins:

extraEnv: |
  - name: AIRFLOW__CORE__LAZY_LOAD_PLUGINS
    value: "False"
  - name: AIRFLOW__WEBSERVER__RELOAD_ON_PLUGIN_CHANGE
    value: "True"

But still, whenever I change the static file:

DUMMY_DAG = "DUMMY_DAG"
...
...
DUMMY_DAG_2 = "DUMMY_DAG_2"

I get the following error on every DAG in my repository:

Broken DAG: [/opt/airflow/dags/repo/dags/ops/dummy_dag.py] Traceback (most recent call last):
  File "/opt/airflow/dags/repo/dags/ops/dummy_dag.py", line 11, in <module>
    from providers.priority_provider import get_schedule
  File "/opt/airflow/dags/repo/plugins/providers/dummy_dag.py", line 170, in <module>
    dag_names.DUMMY_DAG_2
AttributeError: module 'variables.dag_names' has no attribute 'DUMMY_DAG_2'. Did you mean: 'DUMMY_DAG'?

WHYYYYYY

[kobethuwis]

Okay sorry, I misunderstood. Previously, I thought I had to simply rename the folder plugins to a different folder, but you are suggesting I create a subfolder, makes sense 😅 Am trying it rn

[kobethuwis]

@potiuk I tried your suggestion and it still is not working. I now have a combination of the following:

extraEnv:
  - name: AIRFLOW__CORE__LAZY_LOAD_PLUGINS
    value: "False"
  - name: AIRFLOW__WEBSERVER__RELOAD_ON_PLUGIN_CHANGE
    value: "True"

I still get the same error. I tried without the extraEnv values, each one of them and both, to no avail. 😢

[kobethuwis]

Each DAG throws the following error

    from lineas.providers.priority_provider import get_schedule
  File "/opt/airflow/dags/repo/plugins/lineas/providers/priority_provider.py", line 169, in <module>
    dag_names.DUMMY123
AttributeError: module 'lineas.variables.dag_names' has no attribute 'DUMMY123'

Which is on this block of code:

from lineas.variables import dag_names
def get_schedule(dag_name):
    from airflow import AirflowException
    timeslots_list = list(timeslots.keys())

    for ts in timeslots_list:
        for count, dag in enumerate(timeslots.get(ts).get("DAGS")):
            if (isinstance(dag, str) and dag_name == dag) or (
                    isinstance(dag, list) and dag_name in dag):
                schedule = timeslots.get(ts).get('schedule')
                if schedule is not None:
                    return add_minutes_to_schedule(
                        schedule, count * MINUTE_INTERVAL_BETWEEN_SCHEDULES)
                else:
                    return schedule

    for custom_timeslot in custom_timeslots:
        if dag_name == custom_timeslot["dag_name"]:
            return custom_timeslot["schedule"]

    raise AirflowException(f"Dag {dag_name} does not exist")

And our DAGs look like this:

from airflow import DAG
from lineas.variables import dag_owners, dag_names, dag_tags, dag_variables

dag_name = dag_names.DUMMY

custom_args = {
    "owner": dag_owners.KOBE_THUWIS,
    "start_date": datetime(2023, 10, 25),
    "retries": 1
}

dag = DAG(
    dag_name,
    default_args=dag_variables.generate_default_args(custom_args, dag_name),
    schedule_interval=get_schedule(dag_name),
    on_failure_callback=dag_failure_callback
)

[potiuk]

Ah yes. of course. if you want to RELOAD plugins in scheduler you need to restart it. I thought the problem was with variables not with reloading. Scheduler does not have hot plugin reload - only webserver (and DagFileProcessor).

Hot reloading is next to impossible to be done in Python for multiple reasons. The only way it works in Webserver is because it actually has N processes and they kept on being restarted regularly as they are just statelessly handling requests and lazy loading of plugins make them load only after the worker processes start., Similarly DAGFileProcessor forks a new process every time it parses a file so plugins (and any other imported python modiules) are loaded after the process if forked.

But scheduler is a single process that runs in a continuous loop, so if you want to reload Python modules you need to restart the scheduler process yourself.

[kobethuwis]

Okay, long story short: static files do not work for the dag_name & schedule attribute. I think this is due to that the scheduler needs these args to initialize the DAGs.

So this works:

from dag_tags import dag_tags
dag = DAG(
    "my_dag",
    default_args=dag_variables.generate_default_args(custom_args, dag_name),
    schedule_interval=get_schedule("my_dag"),
    tags=[dag_tags.my_dag_tag],
    on_failure_callback=dag_failure_callback
)

but this does not:

from dag_names import dag_names
dag = DAG(
    dag_names.my_dag ,
    default_args=dag_variables.generate_default_args(custom_args, dag_name),
    schedule_interval=get_schedule(dag_names.my_dag),
    tags=[dag_tags.my_dag_tag],
    on_failure_callback=dag_failure_callback
)

I fixed the error by simply adding all of the dag_names in the same file as the get_schedule() plugin folder 🤷‍♂️