Also track task_log_prefix_template changes

[uranusjr]

Extending #20165 (for task #19625) because it turns out tracking just log_filename_template is not enough.

The LogFilename model is renamed to LogTemplate, and now contains two template fields filename (for log_filename_template) and task_prefix (for task_log_prefix_template). The two fields are tracked together when Airflow launches as previously.

The custom task log formatter is updated to use the database entry instead of the configuration, similar to the log filename renderer.

Since the previous change has not been released, I opted to update the migration in-place instead of creating a new one.

Also added some more tests on task_log_prefix_template.

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[potiuk]

Should we turn it into a wanrning and fall-back to the first log template id ?

I think exception on missing log id is a bit too much as it might result in "stack-trace error" (maybe that is intended though?).

Also maybe the message could be somehow more specific. As the current one is not really actionable. I presume at the first start of airlfow there will be at least one entry created, so I guess the only case it can happen is when log_template_id is set in DagRun but missing in the LogTemplate.

Maybe we should be very explicit "Your dag_run table is corrupted. The entry in your dag_run table with dag_run_id: contains invalid log_template_id: <log_template_id>. You should set log_template_id for this field to one of the existing log_template_ids or set to NULL." . Or something like that.

[uranusjr]

This can never happen because 1. None has special meaning, and 2. there’s a foreign key constraint so the database guanratees any non-null value points to something. Things are going very wrong if we reach this code, so I think it’s better to fail extremely loudly.

[uranusjr]

tests/executors/test_dask_executor.py::TestDaskExecutor::test_dask_executor_functions:
ValueError: The futures should have finished; there is probably an error communicating with the Dask cluster.

Hmm. I don’t recall seeing this being flaky previously, maybe something new going wrong? Likely not related though.