Load Example Plugins with Example DAGs

[jscheffl]

I noticed that we have some example plugins we ship in our source code, but so far we are limited not being able to "eat our own dog-food" with the example DAGs. Plugins so far needed to be deployed by deployment manager.

This PR improves the plugin manager slightly in the way that plugins contained in example DAGs folder in source are loaded as well when example DAGs are loaded.

This enables us(in separate PRs in future... soon?) to really use timetables in the example DAGs and not only have code examples linked in documentation.

FYI @hussein-awala I remember you also had a hard time making an example working in our PR #36029

[potiuk]

Some serialization tests are failing but I like it :)

[potiuk]

LGTM. @uranusjr ?

[uranusjr]

Looks good to me, one minor thought.

[zachliu]

Sorry for commenting on a merged PR. Just wanna make sure this is a real issue before i create an issue 😅

I'm on Airflow 2.10.1.

The PR does make sure the plugin is loaded when load_examples = True. And the DAG itself runs ok. No "Broken Dag" import errors or anything. The UI -> Admin -> Plugins shows the 3 plugins from the example_dags/plugins

However, both the sync-perm cli and the plugins cli don't agree with it. When executing airflow sync-perm --include-dags -v, it always errors out on this example dag:

{serialized_dag.py:200} DEBUG - Deserializing DAG: example_workday_timetable
{plugins_manager.py:351} DEBUG - Plugins are already loaded. Skipping.
{plugins_manager.py:484} DEBUG - Initialize extra timetables plugins
{cli_action_loggers.py:98} DEBUG - Calling callbacks: []
Traceback (most recent call last):
File "/usr/local/airflow/.local/bin/airflow", line 8, in <module>
sys.exit(main())
^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/__main__.py", line 62, in main
args.func(args)
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/cli/cli_config.py", line 49, in command
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/utils/cli.py", line 115, in wrapper
return f(*args, **kwargs)
^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/utils/providers_configuration_loader.py", line 55, in wrapped_function
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/fab/auth_manager/cli_commands/sync_perm_command.py", line 39, in sync_perm
appbuilder.sm.create_dag_specific_permissions()
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/fab/auth_manager/security_manager/override.py", line 1072, in create_dag_specific_permissions
dagbag.collect_dags_from_db()
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/dagbag.py", line 628, in collect_dags_from_db
self.dags = SerializedDagModel.read_all_dags()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/utils/session.py", line 97, in wrapper
return func(*args, session=session, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/serialized_dag.py", line 201, in read_all_dags
dag = row.dag
^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/serialized_dag.py", line 235, in dag
return SerializedDAG.from_dict(data)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/serialization/serialized_objects.py", line 1776, in from_dict
return cls.deserialize_dag(serialized_obj["dag"])
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/serialization/serialized_objects.py", line 1704, in deserialize_dag
v = decode_timetable(v)
^^^^^^^^^^^^^^^^^^^
File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/serialization/serialized_objects.py", line 329, in decode_timetable
raise _TimetableNotRegistered(importable_string)
airflow.serialization.serialized_objects._TimetableNotRegistered: Timetable class 'airflow.example_dags.plugins.workday.AfterWorkdayTimetable' is not registered or you have a top level database access that disrupted the session. Please check the airflow best practices documentation.

and when executing airflow plugins -o table -v, the 3 plugins from example_dags/plugins are not in the output table.

of course if I set load_examples = False, the sync-perm and plugin work fine without any errors

the only way that I can run both this example and the cli commands is:

1. copy the source code airflow/example_dags/plugins/workday.py to my own airflow/plugins (in my airflow.cfg, i have plugins_folder = /usr/local/airflow/plugins)
2. re-create the example as one of my own dags but import from my own plugins: from workday import AfterWorkdayTimetable

[jscheffl]

Sorry for commenting on a merged PR. Just wanna make sure this is a real issue before i create an issue 😅

That is okay. And I assume we have not tried the CLI after the change applied with example DAGs being loaded. Can you raise a new bug for this? (related to this PR?)

Did you set "load examples=True" in your config or did you see this error after running with examples loaded once? Because if the CLI runs w/o examples explicitly turned-on but DAGs have been parsed to DB with examples, this might be a config mis-match.

[zachliu]

Sorry for commenting on a merged PR. Just wanna make sure this is a real issue before i create an issue 😅

That is okay. And I assume we have not tried the CLI after the change applied with example DAGs being loaded. Can you raise a new bug for this? (related to this PR?)

Did you set "load examples=True" in your config or did you see this error after running with examples loaded once? Because if the CLI runs w/o examples explicitly turned-on but DAGs have been parsed to DB with examples, this might be a config mis-match.

Yes, I've been setting load_examples=True. I'll create a new issue today.