You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "nickstenning (via GitHub)" <gi...@apache.org> on 2026/02/18 14:35:25 UTC
[PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
nickstenning opened a new pull request, #62128:
URL: https://github.com/apache/airflow/pull/62128
DagFileProcessorManager has been emitting a nonsense value for `dag_processing.total_parse_time` since 8774f28d76, which reversed the order in which `emit_metrics` and `prepare_file_queue` (then called `prepare_file_path_queue`) were called.
As `prepare_file_path_queue` was responsible for resetting the value of `self._parsing_start_time`, the assumption made by `emit_metrics` was that it would be called once the file queue had been cleared, but crucially before `prepare_file_queue` was called to refill the queue.
I've put the order of these calls back to one that generates valid metrics, and restructured the code in order to make the data dependencies explicit so that it's harder to make the same mistake in future refactoring.
<!-- SPDX-License-Identifier: Apache-2.0 https://www.apache.org/licenses/LICENSE-2.0 -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2834683254
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -36,7 +36,12 @@
from datetime import datetime, timedelta
from operator import attrgetter, itemgetter
from pathlib import Path
-from typing import TYPE_CHECKING, Any, NamedTuple, cast
+from typing import TYPE_CHECKING, Any, Literal, NamedTuple, cast
+
+if sys.version_info >= (3, 11):
+ from typing import assert_never
+else:
+ from typing_extensions import assert_never
Review Comment:
You could move this to https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/typing_compat.py#L29-L32
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -36,7 +36,12 @@
from datetime import datetime, timedelta
from operator import attrgetter, itemgetter
from pathlib import Path
-from typing import TYPE_CHECKING, Any, NamedTuple, cast
+from typing import TYPE_CHECKING, Any, Literal, NamedTuple, cast
+
+if sys.version_info >= (3, 11):
+ from typing import assert_never
+else:
+ from typing_extensions import assert_never
Review Comment:
You could move this to https://github.com/apache/airflow/blob/ef5608204c8d039f2ef30f6d78617c0db26ac94c/airflow-core/src/airflow/typing_compat.py#L29-L32
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "nickstenning (via GitHub)" <gi...@apache.org>.
nickstenning commented on PR #62128:
URL: https://github.com/apache/airflow/pull/62128#issuecomment-3937510883
If someone could rerun the tests that would be ace. Looks like a flake to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #62128:
URL: https://github.com/apache/airflow/pull/62128
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2834703829
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1160,13 +1164,34 @@ def _kill_timed_out_processors(self):
processor = self._processors.pop(proc)
processor.logger_filehandle.close()
- def _add_files_to_queue(self, files: list[DagFileInfo], add_at_front: bool):
+ def _add_files_to_queue(
+ self,
+ files: list[DagFileInfo],
+ *,
+ mode: Literal["front", "back", "frontprio"],
+ ):
"""Add stuff to the back or front of the file queue, unless it's already present."""
- new_files = list(f for f in files if f not in self._file_queue)
- if add_at_front:
+ if mode == "frontprio":
+ for file in files:
+ # Try removing the file if already present
+ with contextlib.suppress(ValueError):
+ self._file_queue.remove(file)
+ # enqueue file to the start of the queue.
+ self._file_queue.appendleft(file)
+ elif mode == "front":
+ new_files = list(f for f in files if f not in self._file_queue)
self._file_queue.extendleft(new_files)
- else:
+ elif mode == "back":
+ new_files = list(f for f in files if f not in self._file_queue)
self._file_queue.extend(new_files)
+ else:
+ assert_never(mode)
+
+ # If we've just added files to the queue for the first time since metrics were last emitted, reset the
+ # parse time counter.
+ if not self._parsing_start_time and self._file_queue:
Review Comment:
```suggestion
if self._parsing_start_time is not None and self._file_queue:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "nickstenning (via GitHub)" <gi...@apache.org>.
nickstenning commented on PR #62128:
URL: https://github.com/apache/airflow/pull/62128#issuecomment-3936871320
Thanks for re-review @kaxil. Should be ready to merge once CI passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2822760868
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -393,11 +393,16 @@ def _run_parsing_loop(self):
self._refresh_dag_bundles(known_files=known_files)
if not self._file_queue:
+ if self._parsing_start_time:
Review Comment:
Could you add a simple test:
1) First cycle with empty queue does not emit parse-time metric and does not crash,
2) Subsequent empty-queue transition emits `dag_processing.total_parse_time` from the previous cycle window,
3) `_parsing_start_time` reset semantics are correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2834702516
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1076,7 +1072,15 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
Note this method is only called when the file path queue is empty
"""
- self._parsing_start_time = time.perf_counter()
+ # We only emit metrics after processing all files in the queue. If `self._parsing_start_time` is None
+ # when this method is called, no files have yet been added to the queue so we shouldn't emit metrics.
+ if self._parsing_start_time:
Review Comment:
```suggestion
if self._parsing_start_time is not None:
```
to not have to worry about 0.0 and such
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2834945641
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1160,13 +1164,34 @@ def _kill_timed_out_processors(self):
processor = self._processors.pop(proc)
processor.logger_filehandle.close()
- def _add_files_to_queue(self, files: list[DagFileInfo], add_at_front: bool):
+ def _add_files_to_queue(
+ self,
+ files: list[DagFileInfo],
+ *,
+ mode: Literal["front", "back", "frontprio"],
+ ):
"""Add stuff to the back or front of the file queue, unless it's already present."""
- new_files = list(f for f in files if f not in self._file_queue)
- if add_at_front:
+ if mode == "frontprio":
+ for file in files:
+ # Try removing the file if already present
+ with contextlib.suppress(ValueError):
+ self._file_queue.remove(file)
+ # enqueue file to the start of the queue.
+ self._file_queue.appendleft(file)
+ elif mode == "front":
+ new_files = list(f for f in files if f not in self._file_queue)
self._file_queue.extendleft(new_files)
- else:
+ elif mode == "back":
+ new_files = list(f for f in files if f not in self._file_queue)
self._file_queue.extend(new_files)
+ else:
+ assert_never(mode)
+
+ # If we've just added files to the queue for the first time since metrics were last emitted, reset the
+ # parse time counter.
+ if not self._parsing_start_time and self._file_queue:
Review Comment:
Whoops haha
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "nickstenning (via GitHub)" <gi...@apache.org>.
nickstenning commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2822811387
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -393,11 +393,16 @@ def _run_parsing_loop(self):
self._refresh_dag_bundles(known_files=known_files)
if not self._file_queue:
+ if self._parsing_start_time:
Review Comment:
Ah, sorry, yes. It should be a `hasattr` or, more likely, we should update to have a default of `None` for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "nickstenning (via GitHub)" <gi...@apache.org>.
nickstenning commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2834891715
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1160,13 +1164,34 @@ def _kill_timed_out_processors(self):
processor = self._processors.pop(proc)
processor.logger_filehandle.close()
- def _add_files_to_queue(self, files: list[DagFileInfo], add_at_front: bool):
+ def _add_files_to_queue(
+ self,
+ files: list[DagFileInfo],
+ *,
+ mode: Literal["front", "back", "frontprio"],
+ ):
"""Add stuff to the back or front of the file queue, unless it's already present."""
- new_files = list(f for f in files if f not in self._file_queue)
- if add_at_front:
+ if mode == "frontprio":
+ for file in files:
+ # Try removing the file if already present
+ with contextlib.suppress(ValueError):
+ self._file_queue.remove(file)
+ # enqueue file to the start of the queue.
+ self._file_queue.appendleft(file)
+ elif mode == "front":
+ new_files = list(f for f in files if f not in self._file_queue)
self._file_queue.extendleft(new_files)
- else:
+ elif mode == "back":
+ new_files = list(f for f in files if f not in self._file_queue)
self._file_queue.extend(new_files)
+ else:
+ assert_never(mode)
+
+ # If we've just added files to the queue for the first time since metrics were last emitted, reset the
+ # parse time counter.
+ if not self._parsing_start_time and self._file_queue:
Review Comment:
Yes but without inverting the sense of the condition 😉
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #62128:
URL: https://github.com/apache/airflow/pull/62128#discussion_r2822782383
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -393,11 +393,16 @@ def _run_parsing_loop(self):
self._refresh_dag_bundles(known_files=known_files)
if not self._file_queue:
+ if self._parsing_start_time:
Review Comment:
`self._parsing_start_time` is defined with `init=False`:
https://github.com/apache/airflow/blob/09fc07ddecbe07b3ad8aae7bf1b6dba5115e9857/airflow-core/src/airflow/dag_processing/manager.py#L210
So this will fail because it hasn't been assigned yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix broken `dag_processing.total_parse_time` metric [airflow]
Posted by "nickstenning (via GitHub)" <gi...@apache.org>.
nickstenning commented on PR #62128:
URL: https://github.com/apache/airflow/pull/62128#issuecomment-3935331627
Updated with a test and some slightly more invasive fixes, because it turns out this was even more broken than I thought! @kaxil PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org