AIP-82 Handle trigger serialization

[vincbeck]

Handle serialization of triggers in DAGs as part of AIP-82. With AIP-82 (External event driven scheduling in Airflow) you can define DAGs as below:

trigger = FileTrigger(....)
asset = Asset("<my_queue>", watchers=[AssetWatcher(name="my_file_watcher", trigger=trigger)])
 
with DAG(
    dag_id=DAG_ID,
    schedule=asset,
):
    empty_task = EmptyOperator(task_id="empty_task")
 
    chain(empty_task)

Triggers can be instantiated as part of a DAG. As such, we need to serialize them.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[vincbeck]

I addressed your comments but I also added something new. It was actually a suggestion from @dstandish that I think is a great idea! I introduced the concept of AssetWatcher that is a thin wrapper around trigger. Explanations:

Instead of having:

trigger = FileTrigger(....)
asset = Asset("<my_queue>", watchers=[trigger])

Now it is:

trigger = FileTrigger(....)
asset = Asset("<my_queue>", watchers=[AssetWatcher(name="my_queue_watcher", trigger=trigger)])

AssetWatcher does not add much besides just a name, but I think it is important because, in the future, we might want to surface the relations between assets and triggers in the graph view in the UI. Representing a trigger in the graph can be tricky because triggers have 2 pieces of information: classpath and kwargs. None of them are very suitable for displaying purposes. classpath is debatable but it is anyway not representable of the trigger. Example: you might have an asset with 2 watchers: each one uses the trigger FileTrigger to monitor a different file. Having only the classpath displayed will not help the user to understand which one is which. Hence the name.

[vincbeck]

@uranusjr @ashb what do you guys think?

[ashb]

I'll take another look today or tomorrow.

[ashb]

Looks much better thanks, but a few changes I'd like you to make please.

[ashb]

These casts look odd. What's going on here?

At least this looks like it should work?

Suggested change

	cast(dict, watcher.trigger)["classpath"], cast(dict, watcher.trigger)["kwargs"]
	): cast(dict, watcher.trigger)
	watcher.trigger["classpath"], watcher.trigger["kwargs"]
	): cast(dict, watcher.trigger)

[vincbeck]

The trigger property in AssetWatcher can be a BaseTrigger or a dict depending on where the object AssetWatcher is created:

• In the DAG definition (by users), trigger is a BaseTrigger
• When deserializing the DAG from the DB, trigger is a dict to avoid creating the AssetWatcher as part of the deserialization process (based on your comment and suggestion by TP here)

Here I cast it to make Mypy happy

[ashb]

Did we need to change this?

[vincbeck]

Yes to make one serialization/deserialization test happy. One of the tests serialize and deserialize all the example DAGs and expect the actual example DAGs to equal the serialized -> deserialized one (round trip transformation). Having this made the test fail because file_path was different. I could have mocked it but I also realized that /tmp/ is something widely use as temporary directory in tests so I want with the easy solution :)

Let me know if you want to keep it, I'll have to play with mocks then

[ashb]

Consistency please -- There is another deserialize (why are there two? Separate question I guess) but

Suggested change

	name=var["name"],
	uri=var["uri"],
	group=var["group"],
	extra=var["extra"],
	**var,

[vincbeck]

I dont have the answer for this one. What I know is one if at the root level of deserialize (then deserializing an asset) and the other one is in decode_asset_condition so deserializing an asset condition. I can create a private function to factorize code if you like it better

[ashb]

I don't mind two versions per se, but it'd be nice if they had the same implementation 😁

[vincbeck]

I applied this one but reverted because var contains some keys we do not want to pass to Asset. See error here

[vincbeck]

I factorized the code in a new function decode_asset

[ashb]

I don't think we should make this class do double duty like this -- We already have BaseSerializedOperator etc. and I want to try very hard to avoid putting a run-time dependency on TaskSDK code in the core Scheduler.

i.e. I'd like the scheduler to operate only on a class that is deserialized from the DB that does not inherit from this class.

(I have yet to do that for SerializedDAG and SerializedBaseOperator, but I need to before we can release 3.0)

[vincbeck]

Are you suggesting we should not use this class when we deserialize the DAG? Currently this class is used by users when defining their DAGs and by the scheduler when it deserializes the DAG from the DB. Do you want to use it only by users? I am trying to understand the path forward :)

Do not we have the same problem with classes like Asset?

[ashb]

Do you want to use it only by users? I am trying to understand the path forward :)

That is my end goal yes, and Asset would need updating too. Maybe it's best all done at once. For instance on a SchedulerDag object or a SchedulerBaseOperator (not the final names) the interface will be greatly reduced, and contain only the minimal needed to set relationships between tasks, but a lot of the "richness" of API and compat would not be needed in that class.

But it's probably more confusing to have one class be like that but not others.

That said, one thing you might be able to do here is have a SerializedAssetWatcher class which is a subclass of this one with this field as a different type? This pattern is what we have already for SerializedBaseOperator -- i.e. derserialize code should create instances of SerializedAssetWatcher not AssetWatcher.

If that works with mypy (it might complain about the type of the field not matching the parent) I think this approach is a small change here but makes my life much easier on this bigger refactor when I get to it.

Am I making any sense?

[vincbeck]

You are :) A lot! Thanks a lot for the details, that helps a lot to better understand. I created a new class SerializedAssetWatcher and use that class whenever I deserialize. In collection.py, when I fetch the watchers, I cast them as SerializedAssetWatcher because asset.watchers can be AssetWatcher or SerializedAssetWatcher. Let me know if this is what you had in mind :)