Custom Timetables must be imported relative to $PLUGINS_FOLDER

[Gollum999]

Apache Airflow version

2.2.3 (latest released)

What happened

I wrote a custom Timetable following the example. airflow plugins reports that the plugin is registered correctly, and running the DAG script on the command line reports no errors. But once the webserver attempts to load the dag, it complains:

Broken DAG: [/home/tsanders/airflow/dags/test_airflow2.py] Traceback (most recent call last):
  File "/opt/conda/envs/airflow/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 269, in serialize_to_json
    serialized_object[key] = _encode_timetable(value)
  File "/opt/conda/envs/airflow/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 150, in _encode_timetable
    raise _TimetableNotRegistered(importable_string)
airflow.serialization.serialized_objects._TimetableNotRegistered: Timetable class 'tb.airflow.plugins.live_timetable.LiveCronTimetable' is not registered

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/conda/envs/airflow/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 935, in to_dict
    json_dict = {"__version": cls.SERIALIZER_VERSION, "dag": cls.serialize_dag(var)}
  File "/opt/conda/envs/airflow/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 847, in serialize_dag
    raise SerializationError(f'Failed to serialize DAG {dag.dag_id!r}: {e}')
airflow.exceptions.SerializationError: Failed to serialize DAG 'test_airflow2_7': Timetable class 'tb.airflow.plugins.live_timetable.LiveCronTimetable' is not registered

What you expected to happen

The DAG should import successfully in all contexts.

How to reproduce

I eventually realized that the problem only arises when the Timetable is imported relative to a directory that is not the $PLUGINS_FOLDER. In my case, I have changed core.plugins_folder to another directory that also happens to be on my PYTHONPATH.

Steps to reproduce:

$ mkdir ~/airflow_plugins
$ export PYTHONPATH=~:$PYTHONPATH
$ export AIRFLOW__CORE__PLUGINS_FOLDER=~/airflow_plugins

$ cat ~/airflow_plugins/custom_timetable.py
from airflow.plugins_manager import AirflowPlugin
from airflow.timetables.interval import DeltaDataIntervalTimetable

class CustomTimetable(DeltaDataIntervalTimetable):
    pass

class CustomTimetablePlugin(AirflowPlugin):
    name = "custom_timetable_plugin"
    timetables = [CustomTimetable]

$ cat ~/airflow/dags/dag.py
#!/usr/bin/env python3

from datetime import datetime, timedelta
import sys

from airflow import DAG, plugins_manager
from airflow.operators.dummy import DummyOperator

from airflow_plugins.custom_timetable import CustomTimetable as Bad
from custom_timetable import CustomTimetable as Good

plugins_manager.initialize_timetables_plugins()
print(sys.path)
print(plugins_manager.as_importable_string(Bad))
print(plugins_manager.as_importable_string(Good))
print(plugins_manager.timetable_classes)

with DAG(
        'timetable_example',
        timetable=Bad(timedelta(hours=1)),  # this breaks
        # timetable=Good(timedelta(hours=1)),  # this doesn't
) as dag:
    task = DummyOperator(task_id='dummy', start_date=datetime.today())

$ airflow plugins
name                    | source                             
========================+====================================
custom_timetable_plugin | $PLUGINS_FOLDER/custom_timetable.py

$ ~/airflow/dags/dag.py
['/home/tsanders/airflow/dags', '/home/tsanders', '/opt/conda/envs/airflow/lib/python39.zip', '/opt/conda/envs/airflow/lib/python3.9', '/opt/conda/envs/airflow/lib/python3.9/lib-dynload', '/home/tsanders/.local/lib/python3.9/site-packages', '/opt/conda/envs/airflow/lib/python3.9/site-packages', '/home/tsanders/airflow/config', '/home/tsanders/airflow_plugins']
airflow_plugins.custom_timetable.CustomTimetable
custom_timetable.CustomTimetable
{'custom_timetable.CustomTimetable': <class 'custom_timetable.CustomTimetable'>}

# Load web UI here, 'timetable_example' task will be broken or not depending on which import was used

Operating System

CentOS 7.4

Versions of Apache Airflow Providers

N/A

Deployment

Other

Deployment details

Currently testing with standalone mode.

Anything else

Possibly related to #19869.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[bmoon4]

I think sample code is wrong.

Asterisk (*) is missing in infer_manual_data_interval's parameter

https://airflow.apache.org/docs/apache-airflow/stable/howto/timetable.html#define-scheduling-logic

https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/timetables/base/index.html#airflow.timetables.base.Timetable.infer_manual_data_interval

[uranusjr]

The asterisk is optional in this case. It means run_after can only be called as a keyword argument, while the example code allows the caller to use it as either positional or keyword. But since the function is always only called with keyword arguments in Airflow, both syntax will work. The example code is not wrong.

As for the absolute import issue, unfortunately this is a (somewhat obsecure and annoying) problem in Python. If you add a directory in sys.path via multiple paths, names imported from different sys.path items have different identities.

$ tree
.
`-- plugins
    |-- __init__.py
    `-- mod.py
$ cat plugins/mod.py
class A:
    pass
$ PYTHONPATH=plugins python -q  # Simulate how Airflow loads the plugin directory.
>>> from plugins.mod import A as AFromAbsolute
>>> from mod import A as AFromPlugin
>>> AFromAbsolute == AFromPlugin
False

And since timetable is loaded from plugin, you must import relative to the plugin directory, or somewhere that’s not related to the plugin directory hierarchy, such as site-packages. There are probably hacks we can use to work around this, but this kind of “working above” the Python language rules is not a good idea to me.

[ForeverWintr]

I'm asking why airflow uses isinstance and stuff, rather than defining its own unique identifier that wouldn't be susceptible to this issue.

[potiuk]

The answer is explained by @uranusjr very well here:

There are probably hacks we can use to work around this, but this kind of “working above” the Python language rules is not a good idea to me.

@ForeverWintr -> if you have an idea how to implement a solution which will bypass Python limitation, you are absolutely free to propose a PR and lead it to completion. You can join ~ 2900 contributors and solve this annoying problem for others. That's a fantastic way to contribute back for the free software you get and countless hours of all those thousands of people.

If you have some concrete proposals, you can start a discussion where you explain those as well. Surely if you think isinstance is not needed for that, you looked at the code and analysed it already, I guess?

[ForeverWintr]

@potiuk I did read the post I responded to, and the text you quote doesn't answer my question. A solution that doesn't rely on object identity wouldn't be a hack, or working above the python language rules.

you can start a discussion where you explain those as well.

That's what I thought I was doing. My proposed solution is to use a unique identifier for each plugin. The first step of contributing is to understand the current design, which is why I asked about it.

Surely if you think isinstance is not needed for that, you looked at the code and analysed it already, I guess?

Nope, I'm asking why isinstance is needed/used here.

[potiuk]

Nope, I'm asking why isinstance is needed/used here.

Feel free to dive in the code and find out. If this is just curiosity, then I think it's on you to look at the code and analyse it. If you hae concrete questions to concrete parts of code, I suggest you describe it in detail in another discussion that you start and explain all your findings and possibly propose a solution. This is a great start for you to understand the details.

Remember that this is an open-source projects where anyone (even you) can contribute their time and effort and we heartily encourage you to do so. You will learn a lot by digging in yourself. Unfortunately if people here will go into details to satisfy curiosity of others, then people who actually contribute, will have very little time to do so (contribute). And code is open and it's best to look at and find out - and best to satisfy the curiosity of yours.

Also by actually digging in, analysing and spending time on understanding, you show that you care and that you are willing to spend even more time on contributing and giving back. Which we heartily encourage you to. By doing it you will show that you are ready not only ready to saisfy your couriosity but also ready to help and spend your time on it.

I strongly encourage you to go that path.

[uranusjr]

why airflow uses isinstance and stuff, rather than defining its own unique identifier

I think it boils down to Airflow is implemented in Python, so we use utilities provided by Python itself to do the job. Introducing an Airflow-specific class identifier would get us pass this check, but it does not fundamentally change the class identities, and as long as Airflow depends on Python, something it uses is going to break (and I suspect something would already break if this check is not there). We are building a workflow tool at the end of the day, not an entire language toolchain, and IMO it is more reasonable to use existing language features and expect setups to conform to the language’s constraints.

[shettyrag]

I am trying the workday example plugin given example directory, when i try import it i get an error.

File "/usr/local/lib/python3.8/site-packages/airflow/serialization/serialized_objects.py", line 306, in serialize_to_json
    serialized_object[key] = _encode_timetable(value)
  File "/usr/local/lib/python3.8/site-packages/airflow/serialization/serialized_objects.py", line 175, in _encode_timetable
    raise _TimetableNotRegistered(importable_string)
airflow.serialization.serialized_objects._TimetableNotRegistered: Timetable class 'airflow.example_dags.plugins.workday.AfterWorkdayTimetable' is not registered

any suggestion to reslove it?

[potiuk]

Converting that to a discussion - I guess there was some configuraiton issue (and I if I understand well it's been solved by restarting).