airflow

The feature I am trying to add is a “Dag Stats” column to the main admin view that shows the status of every single run of a dag (running, failed, success). I renamed the current Dag Stats column to Task Stats, which more accurately reflects its information.

I originally accomplished this through a single database query against the dagrun table every time the page is loaded, but Sid pointed out that this would perform incredibly bad for any user with a large number of dags or dag runs.

I’m currently using another table called DagStats that stores dag stat data that can be pulled and immediately displayed by the Dag Stats column, and updating the DagStats table whenever a dagrun’s state changes and causes some rows of the DagStats table to become out of date.

There are two static methods in the DagStat class in models--  set_dirty() and clean_dirty(). Set_dirty() takes one argument, the ID of a dag. It then sets the dirty bit of all rows with that Dag ID to true to indicate these rows are out of date. Clean_dirty() takes no arguments, and updates all dirty rows and pulls in rows for dags that weren’t in Dag Stats but for which dag runs do exist.

I then changed the state column of the DagRun class to a declared attr. This lets me capture the moment whenever code changes the state of a dag run, upon which I then call set_dirty() on the dag_id of that dag run.

I also overwrote the constructor for DagRun to call super.__init__ and then immediately call set_dirty on the new dagrun’s dag_id and set a class variable DagRun.commit_has_new_dagrun to True. I also added an after_commit event listener that will call clean_dirty if DagRun.commit_has_new_dagrun is true.

The current problem I am having is when I run this code and manually create a new dagrun from the dagrun page, the app crashes due to a “AttributeError: 'NoneType' object has no attribute '_enable_transaction_accounting'”.