DAG versioning behavior

[tlegrave]

Hello there,

We recently migrated to Airflow V3 and we already love its new features, well done for all the work!

However we are struggling to understand some behaviors on DAG versioning. Here is our scenario.

We have a DAG that triggers multiple DAGs (~100k). On the trigger DAG we've set wait_for_completion=False, so we end up with the ~100k runs in queued state (we limit concurrency at DAG level to 100 parallel runs).

We then make changes to the DAG code, and pushed them (the change is on a pool name for a task). A new version of the DAG is created, and we could check in the db that the data in serialized_dag table includes the pool change.

At this point, runs going from queued state to running (and then success) show up in the UI that they've run on the latest version. But the task instances kept using the older pool.

We've read different things and conducted different tests, but we're struggling to understand precisely when the DAG version is determined. Is it when the DAG gets queued? Or when it goes to running?

Side question, our pattern of triggering ~100k runs from one trigger seems a bit odd: we have troubles monitoring runs then. I don't think it is optimal with Airflow, do you have any advice? These 100k runs are supposed to run for some weeks.

Thank you for the help,

[EvertonSA]

hi, out of curiosity, do you have 1 instance with more than 100 thousand dags? this is the biggest number I have heard off. I though 10 k was a lot.

[potiuk]

Also to answer your question more preciseley: when you enable versioning in Airflow 3 bundle (which requires Bundle that supports it - for example GitBundle), what happens is that the version of Dag used when task is run is the same as the version available when dag-run got created.

What happens when scheduler prepares new instances to run, it creates dag_runs in the table in a queued state. That DagRun has a reference to "version" of bundle that was in the database at the moment of creating the dag_runs. And those tasks from that DagRun when executed will checkout the version of the code that is linked with that bundle version (in case of GitBundle - they will check-out the commit that DagFileProcessor used last time when parsing that bundle.

[tlegrave]

Amazing! A warning looks fair enough to me. We also ran into this issue, because we had overridden KubernetesPodOperator class to be able to attach a custom object (ConfigProperties) on which we had not implemented the proper serialization.

So we ended up with serialized DAG including the object itself (ConfigProperties objet at 0x10d47f620) = hash that differs = automatic version increment!

[potiuk]

Actually @wjddn279 -> I am not sure if you are handling the case where an object is not serialized (like in the case above) in #59430 so maybe a good idea to add it as well.

[wjddn279]

@potiuk

In that case, it would be difficult to check using static analysis. While it might be possible, I expect there would be few hits since these are often written in separate files.

Instead, we could check whether serialization is performed consistently during the serialization process (using hasattr(config, '__repr__')), and if not present, display a DAG warning in the same manner.

[potiuk]

Yes. Likely that's a better approach.